diff --git a/src/runtime/core/src/runtime.rs b/src/runtime/core/src/runtime.rs index 782001946..1737e5373 100644 --- a/src/runtime/core/src/runtime.rs +++ b/src/runtime/core/src/runtime.rs @@ -768,11 +768,6 @@ impl AgentRuntime { function_id, tracked.function_name, tracked.endpoint ); - // Store the tracking info first (no PyO3 involvement) - self.topology - .llm_providers - .insert(function_id.clone(), tracked.clone()); - // Create LlmProviderInfo and send event let provider_info = LlmProviderInfo { function_id: function_id.clone(), @@ -783,10 +778,30 @@ impl AgentRuntime { vendor: provider.vendor.clone(), kwargs: kwargs_json, }; - let _ = self + + // Send the event BEFORE recording it in topology. If the send + // fails (channel dropped), we leave the function_id out of + // `llm_providers` so the diff gate re-emits on the next + // heartbeat (self-healing) instead of permanently suppressing + // re-emission and leaving the @MeshLlm proxy unbound. + if let Err(e) = self .event_tx .send(MeshEvent::llm_provider_available(provider_info)) - .await; + .await + { + warn!( + "Failed to emit LLM_PROVIDER_AVAILABLE for function '{}': {}; \ + will retry on next heartbeat", + function_id, e + ); + continue; + } + + // Store the tracking info only after a successful send so the + // diff gate treats this provider as "seen". + self.topology + .llm_providers + .insert(function_id.clone(), tracked.clone()); } } } diff --git a/src/runtime/java/mcp-mesh-spring-ai/src/main/java/io/mcpmesh/ai/handlers/AnthropicHandler.java b/src/runtime/java/mcp-mesh-spring-ai/src/main/java/io/mcpmesh/ai/handlers/AnthropicHandler.java index 09824c2c2..c8fb15228 100644 --- a/src/runtime/java/mcp-mesh-spring-ai/src/main/java/io/mcpmesh/ai/handlers/AnthropicHandler.java +++ b/src/runtime/java/mcp-mesh-spring-ai/src/main/java/io/mcpmesh/ai/handlers/AnthropicHandler.java @@ -7,7 +7,6 @@ import org.springframework.ai.chat.messages.AssistantMessage; import org.springframework.ai.chat.messages.Message; import org.springframework.ai.chat.messages.SystemMessage; -import org.springframework.ai.chat.messages.UserMessage; import org.springframework.ai.chat.model.ChatModel; import org.springframework.ai.chat.model.ChatResponse; import org.springframework.ai.chat.prompt.Prompt; @@ -265,25 +264,8 @@ private LlmResponse generateWithToolsAutoExecute( log.debug("Created {} tool callbacks for ChatClient", toolCallbacks.size()); } - // Extract non-system messages for user content - List nonSystemMessages = new ArrayList<>(); - for (Message msg : springMessages) { - if (!(msg instanceof SystemMessage)) { - nonSystemMessages.add(msg); - } - } - - // Build user content from remaining messages - StringBuilder userContent = new StringBuilder(); - for (Message msg : nonSystemMessages) { - if (msg instanceof UserMessage um) { - if (userContent.length() > 0) userContent.append("\n"); - userContent.append(um.getText()); - } else if (msg instanceof AssistantMessage am) { - if (userContent.length() > 0) userContent.append("\n"); - userContent.append("[Previous Assistant Response]\n").append(am.getText()); - } - } + // Build user content from non-system messages + String userContent = buildUserContent(springMessages); // Use ChatClient with tools ChatClient chatClient = ChatClient.create(model); @@ -295,7 +277,7 @@ private LlmResponse generateWithToolsAutoExecute( } // Add user content - requestSpec.user(userContent.toString()); + requestSpec.user(userContent); // Add tools if present - Spring AI handles tool execution automatically if (!toolCallbacks.isEmpty()) { @@ -336,7 +318,7 @@ private LlmResponse generateWithToolsAutoExecute( if (hintSystemPrompt != null && !hintSystemPrompt.isEmpty()) { retrySpec.system(hintSystemPrompt); } - retrySpec.user(userContent.toString()); + retrySpec.user(userContent); if (!toolCallbacks.isEmpty()) { retrySpec.toolCallbacks(toolCallbacks.toArray(new ToolCallback[0])); } @@ -372,22 +354,7 @@ private LlmResponse generateWithToolsNoExecute( List hintTools) { // Replace system message with formatted one - List messagesWithFormattedSystem = new ArrayList<>(); - boolean addedSystem = false; - for (Message msg : springMessages) { - if (msg instanceof SystemMessage) { - if (!addedSystem && formattedSystemPrompt != null && !formattedSystemPrompt.isEmpty()) { - messagesWithFormattedSystem.add(new SystemMessage(formattedSystemPrompt)); - addedSystem = true; - } - } else { - messagesWithFormattedSystem.add(msg); - } - } - // Add system prompt at beginning if not already added - if (!addedSystem && formattedSystemPrompt != null && !formattedSystemPrompt.isEmpty()) { - messagesWithFormattedSystem.add(0, new SystemMessage(formattedSystemPrompt)); - } + List messagesWithFormattedSystem = replaceSystemMessage(springMessages, formattedSystemPrompt); // Create tool callbacks for schema only (no execution) List toolCallbacks = createToolCallbacksForSchema(tools); diff --git a/src/runtime/java/mcp-mesh-spring-ai/src/main/java/io/mcpmesh/ai/handlers/GeminiHandler.java b/src/runtime/java/mcp-mesh-spring-ai/src/main/java/io/mcpmesh/ai/handlers/GeminiHandler.java index 5372d0c7a..766a76a47 100644 --- a/src/runtime/java/mcp-mesh-spring-ai/src/main/java/io/mcpmesh/ai/handlers/GeminiHandler.java +++ b/src/runtime/java/mcp-mesh-spring-ai/src/main/java/io/mcpmesh/ai/handlers/GeminiHandler.java @@ -6,19 +6,16 @@ import org.springframework.ai.chat.messages.AssistantMessage; import org.springframework.ai.chat.messages.Message; import org.springframework.ai.chat.messages.SystemMessage; -import org.springframework.ai.chat.messages.UserMessage; import org.springframework.ai.chat.model.ChatModel; import org.springframework.ai.chat.model.ChatResponse; import org.springframework.ai.chat.prompt.ChatOptions; import org.springframework.ai.chat.prompt.Prompt; import org.springframework.ai.tool.ToolCallback; -import org.springframework.ai.tool.function.FunctionToolCallback; import org.springframework.ai.google.genai.GoogleGenAiChatOptions; import org.springframework.ai.vertexai.gemini.VertexAiGeminiChatModel; import org.springframework.ai.vertexai.gemini.VertexAiGeminiChatOptions; import java.util.*; -import java.util.function.Function; /** * LLM provider handler for Google Gemini models. @@ -55,6 +52,12 @@ public String[] getAliases() { return new String[]{"google"}; } + /** Gemini uses a shorter previous-turn prefix than the Anthropic/OpenAI default. */ + @Override + public String previousResponsePrefix() { + return "[Previous Response]\n"; + } + // ========================================================================= // Structured Output Methods // ========================================================================= @@ -185,25 +188,8 @@ private LlmResponse generateWithToolsAutoExecute( log.debug("Created {} tool callbacks for ChatClient", toolCallbacks.size()); } - // Extract non-system messages - List nonSystemMessages = new ArrayList<>(); - for (Message msg : springMessages) { - if (!(msg instanceof SystemMessage)) { - nonSystemMessages.add(msg); - } - } - - // Build user content from remaining messages - StringBuilder userContent = new StringBuilder(); - for (Message msg : nonSystemMessages) { - if (msg instanceof UserMessage um) { - if (userContent.length() > 0) userContent.append("\n"); - userContent.append(um.getText()); - } else if (msg instanceof AssistantMessage am) { - if (userContent.length() > 0) userContent.append("\n"); - userContent.append("[Previous Response]\n").append(am.getText()); - } - } + // Build user content from non-system messages + String userContent = buildUserContent(springMessages); // Use ChatClient with tools ChatClient chatClient = ChatClient.create(model); @@ -215,7 +201,7 @@ private LlmResponse generateWithToolsAutoExecute( } // Add user content - requestSpec.user(userContent.toString()); + requestSpec.user(userContent); // Add tools if present if (!toolCallbacks.isEmpty()) { @@ -251,22 +237,7 @@ private LlmResponse generateWithToolsNoExecute( OutputSchema outputSchema) { // Replace system message with formatted one - List messagesWithFormattedSystem = new ArrayList<>(); - boolean addedSystem = false; - for (Message msg : springMessages) { - if (msg instanceof SystemMessage) { - if (!addedSystem && formattedSystemPrompt != null && !formattedSystemPrompt.isEmpty()) { - messagesWithFormattedSystem.add(new SystemMessage(formattedSystemPrompt)); - addedSystem = true; - } - } else { - messagesWithFormattedSystem.add(msg); - } - } - // Add system prompt at beginning if not already added - if (!addedSystem && formattedSystemPrompt != null && !formattedSystemPrompt.isEmpty()) { - messagesWithFormattedSystem.add(0, new SystemMessage(formattedSystemPrompt)); - } + List messagesWithFormattedSystem = replaceSystemMessage(springMessages, formattedSystemPrompt); // Create tool callbacks for schema only (no execution) List toolCallbacks = createToolCallbacksForSchema(tools); @@ -432,49 +403,12 @@ private void applyResponseFormat(ChatClient.ChatClientRequestSpec requestSpec, O } /** - * Create ToolCallbacks for schema only (no execution). - * - *

Overrides the default to apply Gemini-specific uppercase type conversion. + * Apply Gemini-specific uppercase type conversion to a tool's input schema + * before it is serialized and attached to the Spring AI tool callback. */ @Override - public List createToolCallbacksForSchema(List tools) { - List callbacks = new ArrayList<>(); - if (tools == null) return callbacks; - - for (ToolDefinition tool : tools) { - // Create a dummy function that should never be called - java.util.function.Function, String> dummyFunction = args -> { - log.warn("Tool {} was unexpectedly called - this shouldn't happen!", tool.name()); - return "{\"error\": \"Tool execution not supported in provider mode\"}"; - }; - - // Convert inputSchema Map to JSON string with uppercase types for Gemini - String inputSchemaJson = null; - if (tool.inputSchema() != null && !tool.inputSchema().isEmpty()) { - try { - Map convertedSchema = convertSchemaTypesToUpperCase(tool.inputSchema()); - inputSchemaJson = MAPPER - .writeValueAsString(convertedSchema); - log.debug("Converted tool schema for {}: {}", tool.name(), inputSchemaJson); - } catch (Exception e) { - log.warn("Failed to serialize inputSchema for {}: {}", tool.name(), e.getMessage()); - } - } - - @SuppressWarnings("unchecked") - var builder = FunctionToolCallback - .builder(tool.name(), dummyFunction) - .description(tool.description() != null ? tool.description() : "No description") - .inputType((Class>) (Class) Map.class); - - if (inputSchemaJson != null) { - builder.inputSchema(inputSchemaJson); - } - - callbacks.add(builder.build()); - } - - return callbacks; + public Map transformToolInputSchema(Map schema) { + return convertSchemaTypesToUpperCase(schema); } /** @@ -557,53 +491,6 @@ private Map convertSchemaTypesToUpperCase(Map sc return result; } - /** - * Create a Spring AI ToolCallback from our ToolDefinition. - * - *

Overrides the default to apply Gemini-specific uppercase type conversion. - */ - @Override - public ToolCallback createToolCallback(ToolDefinition tool, ToolExecutorCallback toolExecutor) { - Function, String> toolFunction = args -> { - try { - String argsJson = args != null ? MAPPER - .writeValueAsString(args) : "{}"; - return toolExecutor.execute(tool.name(), argsJson); - } catch (Exception e) { - log.error("Tool execution failed: {}", tool.name(), e); - try { - return TOOL_CALLBACK_MAPPER.writeValueAsString(Map.of("error", "Tool execution failed: " + tool.name())); - } catch (Exception ignored) { - return "{\"error\": \"tool execution failed\"}"; - } - } - }; - - // Convert inputSchema with uppercase types for Gemini API compatibility - String inputSchemaJson = null; - if (tool.inputSchema() != null && !tool.inputSchema().isEmpty()) { - try { - Map convertedSchema = convertSchemaTypesToUpperCase(tool.inputSchema()); - inputSchemaJson = MAPPER - .writeValueAsString(convertedSchema); - } catch (Exception e) { - log.warn("Failed to serialize inputSchema for {}: {}", tool.name(), e.getMessage()); - } - } - - @SuppressWarnings("unchecked") - var builder = FunctionToolCallback - .builder(tool.name(), toolFunction) - .description(tool.description() != null ? tool.description() : "No description") - .inputType((Class>) (Class) Map.class); - - if (inputSchemaJson != null) { - builder.inputSchema(inputSchemaJson); - } - - return builder.build(); - } - @Override public Map getCapabilities() { return Map.of( diff --git a/src/runtime/java/mcp-mesh-spring-ai/src/main/java/io/mcpmesh/ai/handlers/LlmProviderHandler.java b/src/runtime/java/mcp-mesh-spring-ai/src/main/java/io/mcpmesh/ai/handlers/LlmProviderHandler.java index a3536ca96..3db210fbc 100644 --- a/src/runtime/java/mcp-mesh-spring-ai/src/main/java/io/mcpmesh/ai/handlers/LlmProviderHandler.java +++ b/src/runtime/java/mcp-mesh-spring-ai/src/main/java/io/mcpmesh/ai/handlers/LlmProviderHandler.java @@ -1,6 +1,10 @@ package io.mcpmesh.ai.handlers; import org.slf4j.LoggerFactory; +import org.springframework.ai.chat.messages.AssistantMessage; +import org.springframework.ai.chat.messages.Message; +import org.springframework.ai.chat.messages.SystemMessage; +import org.springframework.ai.chat.messages.UserMessage; import org.springframework.ai.chat.model.ChatModel; import org.springframework.ai.chat.model.ChatResponse; import org.springframework.ai.tool.ToolCallback; @@ -305,6 +309,94 @@ static String outputModeOverride(Map options) { return v != null ? v.toString() : OUTPUT_MODE_UNSET; } + // ========================================================================= + // Shared Message Assembly + // ========================================================================= + + /** + * Prefix prepended to a previous assistant turn when flattening the message + * history into a single user-content string. + * + *

Vendors override this where their convention differs (e.g. Gemini uses + * {@code "[Previous Response]\n"}). + * + * @return The assistant-turn prefix literal (including its trailing newline) + */ + default String previousResponsePrefix() { + return "[Previous Assistant Response]\n"; + } + + /** + * Extract the non-system messages from a Spring AI message list, preserving order. + * + * @param messages The full message list (may contain system messages) + * @return A new list containing only the non-system messages + */ + default List extractNonSystemMessages(List messages) { + List nonSystemMessages = new ArrayList<>(); + for (Message msg : messages) { + if (!(msg instanceof SystemMessage)) { + nonSystemMessages.add(msg); + } + } + return nonSystemMessages; + } + + /** + * Flatten the non-system messages into a single user-content string. + * + *

User turns are appended verbatim; assistant turns are prefixed with + * {@link #previousResponsePrefix()}. Turns are newline-joined. + * + * @param messages The message list (system messages are ignored) + * @return The assembled user-content string + */ + default String buildUserContent(List messages) { + StringBuilder userContent = new StringBuilder(); + for (Message msg : extractNonSystemMessages(messages)) { + if (msg instanceof UserMessage um) { + if (userContent.length() > 0) userContent.append("\n"); + userContent.append(um.getText()); + } else if (msg instanceof AssistantMessage am) { + if (userContent.length() > 0) userContent.append("\n"); + userContent.append(previousResponsePrefix()).append(am.getText()); + } + } + return userContent.toString(); + } + + /** + * Replace the first system message with the formatted system prompt, dropping + * any additional system messages. If the input contained no system message, + * the formatted prompt is prepended. + * + *

No-op replacement occurs when {@code formattedSystemPrompt} is null/empty: + * any existing system messages are simply dropped and nothing is prepended. + * + * @param messages The full message list + * @param formattedSystemPrompt The formatted system prompt (may be null/empty) + * @return A new list with the system message replaced/prepended + */ + default List replaceSystemMessage(List messages, String formattedSystemPrompt) { + List messagesWithFormattedSystem = new ArrayList<>(); + boolean addedSystem = false; + for (Message msg : messages) { + if (msg instanceof SystemMessage) { + if (!addedSystem && formattedSystemPrompt != null && !formattedSystemPrompt.isEmpty()) { + messagesWithFormattedSystem.add(new SystemMessage(formattedSystemPrompt)); + addedSystem = true; + } + } else { + messagesWithFormattedSystem.add(msg); + } + } + // Add system prompt at beginning if not already added + if (!addedSystem && formattedSystemPrompt != null && !formattedSystemPrompt.isEmpty()) { + messagesWithFormattedSystem.add(0, new SystemMessage(formattedSystemPrompt)); + } + return messagesWithFormattedSystem; + } + // ========================================================================= // Capability Methods // ========================================================================= @@ -340,6 +432,13 @@ default String[] getAliases() { /** Shared ObjectMapper for tool callback serialization. */ static final tools.jackson.databind.ObjectMapper TOOL_CALLBACK_MAPPER = new tools.jackson.databind.ObjectMapper(); + /** Vendor hook to transform a tool's input JSON schema before it is attached + * to the Spring AI tool callback. Default: identity. Gemini overrides to + * upper-case JSON-schema type values. */ + default Map transformToolInputSchema(Map schema) { + return schema; + } + /** * Create ToolCallbacks for schema only (no execution). * @@ -360,7 +459,7 @@ default List createToolCallbacksForSchema(List too String inputSchemaJson = null; if (tool.inputSchema() != null && !tool.inputSchema().isEmpty()) { try { - inputSchemaJson = TOOL_CALLBACK_MAPPER.writeValueAsString(tool.inputSchema()); + inputSchemaJson = TOOL_CALLBACK_MAPPER.writeValueAsString(transformToolInputSchema(tool.inputSchema())); } catch (Exception e) { LoggerFactory.getLogger(getClass()).warn("Failed to serialize inputSchema for {}: {}", tool.name(), e.getMessage()); } @@ -407,7 +506,7 @@ default ToolCallback createToolCallback(ToolDefinition tool, ToolExecutorCallbac String inputSchemaJson = null; if (tool.inputSchema() != null && !tool.inputSchema().isEmpty()) { try { - inputSchemaJson = TOOL_CALLBACK_MAPPER.writeValueAsString(tool.inputSchema()); + inputSchemaJson = TOOL_CALLBACK_MAPPER.writeValueAsString(transformToolInputSchema(tool.inputSchema())); } catch (Exception e) { LoggerFactory.getLogger(getClass()).warn("Failed to serialize inputSchema for {}: {}", tool.name(), e.getMessage()); } diff --git a/src/runtime/java/mcp-mesh-spring-ai/src/main/java/io/mcpmesh/ai/handlers/OpenAiHandler.java b/src/runtime/java/mcp-mesh-spring-ai/src/main/java/io/mcpmesh/ai/handlers/OpenAiHandler.java index ca9f75162..09f063bc3 100644 --- a/src/runtime/java/mcp-mesh-spring-ai/src/main/java/io/mcpmesh/ai/handlers/OpenAiHandler.java +++ b/src/runtime/java/mcp-mesh-spring-ai/src/main/java/io/mcpmesh/ai/handlers/OpenAiHandler.java @@ -6,7 +6,6 @@ import org.springframework.ai.chat.messages.AssistantMessage; import org.springframework.ai.chat.messages.Message; import org.springframework.ai.chat.messages.SystemMessage; -import org.springframework.ai.chat.messages.UserMessage; import org.springframework.ai.chat.model.ChatModel; import org.springframework.ai.chat.model.ChatResponse; import org.springframework.ai.chat.prompt.Prompt; @@ -164,25 +163,8 @@ private LlmResponse generateWithToolsAutoExecute( log.debug("Created {} tool callbacks for ChatClient", toolCallbacks.size()); } - // Extract non-system messages for user content - List nonSystemMessages = new ArrayList<>(); - for (Message msg : springMessages) { - if (!(msg instanceof SystemMessage)) { - nonSystemMessages.add(msg); - } - } - - // Build user content from remaining messages - StringBuilder userContent = new StringBuilder(); - for (Message msg : nonSystemMessages) { - if (msg instanceof UserMessage um) { - if (userContent.length() > 0) userContent.append("\n"); - userContent.append(um.getText()); - } else if (msg instanceof AssistantMessage am) { - if (userContent.length() > 0) userContent.append("\n"); - userContent.append("[Previous Assistant Response]\n").append(am.getText()); - } - } + // Build user content from non-system messages + String userContent = buildUserContent(springMessages); // Use ChatClient with tools ChatClient chatClient = ChatClient.create(model); @@ -194,7 +176,7 @@ private LlmResponse generateWithToolsAutoExecute( } // Add user content - requestSpec.user(userContent.toString()); + requestSpec.user(userContent); // Add tools if present - Spring AI handles tool execution automatically if (!toolCallbacks.isEmpty()) { @@ -253,22 +235,7 @@ private LlmResponse generateWithToolsNoExecute( OutputSchema outputSchema) { // Replace system message with formatted one - List messagesWithFormattedSystem = new ArrayList<>(); - boolean addedSystem = false; - for (Message msg : springMessages) { - if (msg instanceof SystemMessage) { - if (!addedSystem && formattedSystemPrompt != null && !formattedSystemPrompt.isEmpty()) { - messagesWithFormattedSystem.add(new SystemMessage(formattedSystemPrompt)); - addedSystem = true; - } - } else { - messagesWithFormattedSystem.add(msg); - } - } - // Add system prompt at beginning if not already added - if (!addedSystem && formattedSystemPrompt != null && !formattedSystemPrompt.isEmpty()) { - messagesWithFormattedSystem.add(0, new SystemMessage(formattedSystemPrompt)); - } + List messagesWithFormattedSystem = replaceSystemMessage(springMessages, formattedSystemPrompt); // Create tool callbacks for schema only (no execution) List toolCallbacks = createToolCallbacksForSchema(tools); diff --git a/src/runtime/java/mcp-mesh-spring-boot-starter/src/main/java/io/mcpmesh/spring/MeshLlmAgentProxy.java b/src/runtime/java/mcp-mesh-spring-boot-starter/src/main/java/io/mcpmesh/spring/MeshLlmAgentProxy.java index e69cac051..24f25baa5 100644 --- a/src/runtime/java/mcp-mesh-spring-boot-starter/src/main/java/io/mcpmesh/spring/MeshLlmAgentProxy.java +++ b/src/runtime/java/mcp-mesh-spring-boot-starter/src/main/java/io/mcpmesh/spring/MeshLlmAgentProxy.java @@ -2,8 +2,10 @@ import tools.jackson.core.JacksonException; import tools.jackson.core.type.TypeReference; +import tools.jackson.databind.DeserializationFeature; import tools.jackson.databind.JsonNode; import tools.jackson.databind.ObjectMapper; +import tools.jackson.databind.json.JsonMapper; import io.mcpmesh.MeshLlmDefaults; import io.mcpmesh.core.MeshObjectMappers; import io.mcpmesh.core.MeshEvent; @@ -59,6 +61,22 @@ public class MeshLlmAgentProxy implements MeshLlmAgent { private static final Logger log = LoggerFactory.getLogger(MeshLlmAgentProxy.class); private static final ObjectMapper objectMapper = MeshObjectMappers.create(); + /** + * Lenient mapper used ONLY for deserializing structured-output responses into a + * user-supplied response model (the {@code generate(Class)} path). + * + *

Under {@code output_mode=hint}, the provider embeds the response schema in the + * prompt but does not enforce it natively, so the LLM may emit loosely-shaped JSON — + * most commonly a scalar where the schema declares a list (e.g. {@code "insights": "x"} + * instead of {@code ["x"]}). {@link DeserializationFeature#ACCEPT_SINGLE_VALUE_AS_ARRAY} + * coerces that single-value-as-array drift. This is a no-op for well-shaped (strict) + * output and is intentionally scoped to response-model parsing only — it must NOT be + * applied to the wire/tool-callback mappers above. + */ + private static final ObjectMapper responseModelMapper = JsonMapper.builder() + .enable(DeserializationFeature.ACCEPT_SINGLE_VALUE_AS_ARRAY) + .build(); + private final String functionId; private final List availableTools = new CopyOnWriteArrayList<>(); private final AtomicReference providerRef = new AtomicReference<>(); @@ -478,12 +496,12 @@ public T generate(Class responseType) { String response = generate(); try { - return objectMapper.readValue(response, responseType); + return responseModelMapper.readValue(response, responseType); } catch (JacksonException e) { String jsonContent = extractJsonFromResponse(response); if (jsonContent != null) { try { - return objectMapper.readValue(jsonContent, responseType); + return responseModelMapper.readValue(jsonContent, responseType); } catch (JacksonException e2) { log.warn("Failed to parse extracted JSON: {}", e2.getMessage()); } diff --git a/src/runtime/java/mcp-mesh-spring-boot-starter/src/test/java/io/mcpmesh/spring/MeshLlmAgentProxyResponseModelLenientTest.java b/src/runtime/java/mcp-mesh-spring-boot-starter/src/test/java/io/mcpmesh/spring/MeshLlmAgentProxyResponseModelLenientTest.java new file mode 100644 index 000000000..68f4f8a37 --- /dev/null +++ b/src/runtime/java/mcp-mesh-spring-boot-starter/src/test/java/io/mcpmesh/spring/MeshLlmAgentProxyResponseModelLenientTest.java @@ -0,0 +1,62 @@ +package io.mcpmesh.spring; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import tools.jackson.databind.ObjectMapper; + +import java.lang.reflect.Field; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Tests that the response-model deserialization path in {@link MeshLlmAgentProxy} + * tolerates the loosely-shaped JSON that {@code output_mode=hint} can produce — + * specifically the documented single-value-as-array drift where the LLM returns a + * scalar for a schema field declared as a list (issue #1142). + * + *

Under hint mode the provider embeds the schema in the prompt but does not + * enforce it natively, so a {@code List} field can come back as a bare + * string ({@code "insights": "x"} instead of {@code ["x"]}). Before the fix this + * threw Jackson {@code MismatchedInputException}. The fix enables + * {@code ACCEPT_SINGLE_VALUE_AS_ARRAY} on a dedicated lenient mapper scoped to the + * {@code generate(Class)} response-model path only. + * + *

The test reflects out the private static {@code responseModelMapper} so it + * exercises the EXACT mapper the proxy uses, not a hand-rolled copy. + */ +class MeshLlmAgentProxyResponseModelLenientTest { + + /** Mirrors a typical @MeshLlm response model with a list field. */ + record Analysis(String summary, List insights) {} + + private static ObjectMapper responseModelMapper() throws Exception { + Field f = MeshLlmAgentProxy.class.getDeclaredField("responseModelMapper"); + f.setAccessible(true); + return (ObjectMapper) f.get(null); + } + + @Test + @DisplayName("scalar string for a List field deserializes to a single-element list (hint-mode drift)") + void scalarCoercedToSingleElementList() throws Exception { + ObjectMapper mapper = responseModelMapper(); + String json = "{\"summary\":\"ok\",\"insights\":\"only-one\"}"; + + Analysis result = mapper.readValue(json, Analysis.class); + + assertEquals("ok", result.summary()); + assertEquals(List.of("only-one"), result.insights()); + } + + @Test + @DisplayName("well-shaped array still deserializes correctly (no-op for strict output)") + void normalArrayStillWorks() throws Exception { + ObjectMapper mapper = responseModelMapper(); + String json = "{\"summary\":\"ok\",\"insights\":[\"a\",\"b\"]}"; + + Analysis result = mapper.readValue(json, Analysis.class); + + assertEquals("ok", result.summary()); + assertEquals(List.of("a", "b"), result.insights()); + } +} diff --git a/src/runtime/python/_mcp_mesh/engine/response_parser.py b/src/runtime/python/_mcp_mesh/engine/response_parser.py index a3aef1d9b..ebe6dd3c4 100644 --- a/src/runtime/python/_mcp_mesh/engine/response_parser.py +++ b/src/runtime/python/_mcp_mesh/engine/response_parser.py @@ -7,7 +7,8 @@ import json import logging -from typing import Any, TypeVar +import typing +from typing import Any, TypeVar, Union import mcp_mesh_core from pydantic import BaseModel, ValidationError @@ -149,6 +150,55 @@ def _parse_json_with_fallback(content: str, output_type: type[T]) -> dict[str, A # If wrapping doesn't work, raise the original JSON error raise ResponseParseError(f"Invalid JSON response: {e}") + @staticmethod + def _is_list_annotation(annotation: Any) -> bool: + """ + Return True if the field annotation is (or optionally wraps) a list/sequence. + + Handles ``list[...]``, ``List[...]`` and ``Optional[list[...]]`` / + ``Union[list[...], None]``. Conservative by design: only annotations that + clearly denote a list participate in scalar-to-array coercion. + """ + origin = typing.get_origin(annotation) + if origin is list: + return True + if origin is Union: + return any( + ResponseParser._is_list_annotation(arg) + for arg in typing.get_args(annotation) + if arg is not type(None) + ) + return False + + @staticmethod + def _coerce_scalar_list_fields( + response_data: dict[str, Any], output_type: type[T] + ) -> dict[str, Any]: + """ + Wrap scalar values in a single-element list for list-typed model fields. + + Scoped to structured-output response-model parsing. Only fields whose + annotation is a list (see :meth:`_is_list_annotation`) and whose received + value is a non-list, non-None scalar are coerced. All other values pass + through unchanged, so well-shaped (strict) output is unaffected. + """ + coerced: dict[str, Any] | None = None + for field_name, field_info in output_type.model_fields.items(): + if field_name not in response_data: + continue + value = response_data[field_name] + if value is None or isinstance(value, (list, tuple)): + continue + if ResponseParser._is_list_annotation(field_info.annotation): + if coerced is None: + coerced = dict(response_data) + logger.debug( + f"📦 Coercing scalar to single-element list for " + f"'{field_name}' (hint-mode drift)" + ) + coerced[field_name] = [value] + return coerced if coerced is not None else response_data + @staticmethod def _validate_and_create(response_data: Any, output_type: type[T]) -> T: """ @@ -227,6 +277,17 @@ def _validate_and_create(response_data: Any, output_type: type[T]) -> T: ) response_data = sole_value + # Single-value-as-array leniency for hint-mode drift (issue #1142). + # Under output_mode=hint the provider embeds the schema in the prompt + # but does not enforce it natively, so the LLM can emit a scalar where + # the schema declares a list (e.g. "insights": "x" instead of ["x"]). + # Wrap such scalars in a single-element list before Pydantic validation. + # No-op for well-shaped (strict) output where the value is already a list. + if isinstance(response_data, dict): + response_data = ResponseParser._coerce_scalar_list_fields( + response_data, output_type + ) + parsed = output_type(**response_data) logger.debug(f"✅ Response parsed successfully: {parsed}") return parsed diff --git a/src/runtime/python/tests/unit/test_response_parser_scalar_list.py b/src/runtime/python/tests/unit/test_response_parser_scalar_list.py new file mode 100644 index 000000000..38dcec093 --- /dev/null +++ b/src/runtime/python/tests/unit/test_response_parser_scalar_list.py @@ -0,0 +1,64 @@ +""" +Unit tests for ResponseParser single-value-as-array leniency (issue #1142). + +Under output_mode=hint the provider embeds the response schema in the prompt +but does not enforce it natively, so the LLM can emit a scalar where the schema +declares a list (e.g. "insights": "x" instead of ["x"]). The parser coerces that +single-value-as-array drift before Pydantic validation. This is a no-op for +well-shaped (strict) output where the value is already a list. +""" + +from typing import List, Optional + +from pydantic import BaseModel + +from _mcp_mesh.engine.response_parser import ResponseParser + + +class Analysis(BaseModel): + summary: str + insights: List[str] + + +class OptionalListModel(BaseModel): + tags: Optional[List[str]] = None + + +class TestResponseParserScalarList: + def test_scalar_coerced_to_single_element_list(self): + """A bare string for a List[str] field becomes a single-element list.""" + data = {"summary": "ok", "insights": "only-one"} + parsed = ResponseParser.parse(data, Analysis) + assert isinstance(parsed, Analysis) + assert parsed.summary == "ok" + assert parsed.insights == ["only-one"] + + def test_scalar_coerced_from_json_string(self): + """Same drift arriving as a raw JSON string is coerced too.""" + content = '{"summary": "ok", "insights": "only-one"}' + parsed = ResponseParser.parse(content, Analysis) + assert parsed.insights == ["only-one"] + + def test_well_shaped_array_unchanged(self): + """A correct array passes through untouched (no-op for strict output).""" + data = {"summary": "ok", "insights": ["a", "b"]} + parsed = ResponseParser.parse(data, Analysis) + assert parsed.insights == ["a", "b"] + + def test_non_list_scalar_field_unaffected(self): + """Scalar fields keep their scalar value; only list fields are coerced.""" + data = {"summary": "ok", "insights": ["a"]} + parsed = ResponseParser.parse(data, Analysis) + assert parsed.summary == "ok" + + def test_optional_list_scalar_coerced(self): + """Optional[List[str]] also coerces a scalar to a single-element list.""" + data = {"tags": "urgent"} + parsed = ResponseParser.parse(data, OptionalListModel) + assert parsed.tags == ["urgent"] + + def test_optional_list_none_unchanged(self): + """A None value for an optional list field is left as None.""" + data = {"tags": None} + parsed = ResponseParser.parse(data, OptionalListModel) + assert parsed.tags is None diff --git a/src/runtime/typescript/src/__tests__/response-parser.test.ts b/src/runtime/typescript/src/__tests__/response-parser.test.ts index 9eb56d36b..f415446db 100644 --- a/src/runtime/typescript/src/__tests__/response-parser.test.ts +++ b/src/runtime/typescript/src/__tests__/response-parser.test.ts @@ -459,3 +459,37 @@ describe("ResponseParseError", () => { expect(error.name).toBe("ResponseParseError"); }); }); + +describe("single-value-as-array leniency (hint-mode drift, #1142)", () => { + const schema = z.object({ + summary: z.string(), + insights: z.array(z.string()), + }); + + it("should coerce a scalar string into a single-element array", () => { + const parser = new ResponseParser(schema); + const result = parser.parse('{"summary": "ok", "insights": "only-one"}'); + expect(result).toEqual({ summary: "ok", insights: ["only-one"] }); + }); + + it("should leave a well-shaped array unchanged (no-op for strict output)", () => { + const parser = new ResponseParser(schema); + const result = parser.parse('{"summary": "ok", "insights": ["a", "b"]}'); + expect(result).toEqual({ summary: "ok", insights: ["a", "b"] }); + }); + + it("should coerce a scalar for an optional array field", () => { + const optionalSchema = z.object({ + tags: z.array(z.string()).optional(), + }); + const parser = new ResponseParser(optionalSchema); + const result = parser.parse('{"tags": "urgent"}'); + expect(result).toEqual({ tags: ["urgent"] }); + }); + + it("should not touch non-array scalar fields", () => { + const parser = new ResponseParser(schema); + const result = parser.parse('{"summary": "ok", "insights": ["a"]}'); + expect(result).toEqual({ summary: "ok", insights: ["a"] }); + }); +}); diff --git a/src/runtime/typescript/src/response-parser.ts b/src/runtime/typescript/src/response-parser.ts index e016d23f2..8bc7f0505 100644 --- a/src/runtime/typescript/src/response-parser.ts +++ b/src/runtime/typescript/src/response-parser.ts @@ -92,6 +92,14 @@ export class ResponseParser { ); } + // Single-value-as-array leniency for hint-mode drift (issue #1142). + // Under output_mode=hint the provider embeds the schema in the prompt but + // does not enforce it natively, so the LLM can emit a scalar where the schema + // declares an array (e.g. "insights": "x" instead of ["x"]). Wrap such scalars + // in a single-element array before Zod validation. No-op for well-shaped + // (strict) output where the value is already an array. + parsed = this.coerceScalarArrayFields(parsed); + // Validate with Zod schema const result = this.schema.safeParse(parsed); if (!result.success) { @@ -108,6 +116,60 @@ export class ResponseParser { return result.data; } + /** + * Wrap scalar values in a single-element array for array-typed schema fields. + * + * Scoped to structured-output response-model parsing. Derives the JSON Schema + * from the Zod schema and, for any top-level property of type `array` whose + * received value is a non-array, non-null scalar, wraps it in `[value]`. All + * other values pass through unchanged, so well-shaped (strict) output where the + * value is already an array is unaffected. + */ + private coerceScalarArrayFields(parsed: unknown): unknown { + if ( + parsed === null || + typeof parsed !== "object" || + Array.isArray(parsed) || + !this.schema + ) { + return parsed; + } + + let jsonSchema: Record; + try { + jsonSchema = zodToJsonSchema(this.schema, { + $refStrategy: "none", + }) as Record; + } catch { + // If schema introspection fails, leave the data untouched and let Zod decide. + return parsed; + } + + const properties = jsonSchema.properties as + | Record + | undefined; + if (!properties) { + return parsed; + } + + const obj = parsed as Record; + let result: Record | null = null; + for (const [key, prop] of Object.entries(properties)) { + if (prop?.type !== "array") continue; + if (!(key in obj)) continue; + const value = obj[key]; + if (value === null || value === undefined || Array.isArray(value)) { + continue; + } + if (result === null) { + result = { ...obj }; + } + result[key] = [value]; + } + + return result ?? parsed; + } + /** * Try to parse, returning null on failure instead of throwing. * diff --git a/tests/integration/suites/uc21_meshjob/tc26_cancel_posts_synthetic_event/test.yaml b/tests/integration/suites/uc21_meshjob/tc26_cancel_posts_synthetic_event/test.yaml index c561acd6e..22ad51364 100644 --- a/tests/integration/suites/uc21_meshjob/tc26_cancel_posts_synthetic_event/test.yaml +++ b/tests/integration/suites/uc21_meshjob/tc26_cancel_posts_synthetic_event/test.yaml @@ -90,20 +90,31 @@ test: capture: commission_response timeout: 60 - # Allow a moment for the producer's stdout to flush after the - # cancelled event arrives. The producer's marker line is logged - # inside the recv_event branch and the runtime captures stdout to - # the agent's log file. - - name: "Wait for producer log to flush" - handler: wait - seconds: 3 - - - name: "Read producer log for the cancelled_gracefully marker" + # Poll the producer's agent log until the cancelled_gracefully marker + # appears. The producer's marker line is logged inside the recv_event + # branch and the runtime captures stdout to the agent's log file + # ASYNCHRONOUSLY — under CI load a fixed wait is sometimes too short + # (issue #1134). Poll up to ~30s, capturing the matching marker line + # as soon as it is present; on timeout dump logs and fail loud. + - name: "Poll producer log for the cancelled_gracefully marker" handler: shell workdir: /workspace command: | - meshctl logs long-task-provider 2>&1 | tail -200 | grep "\[run_until_cancel\] cancelled_gracefully" || true + MARKER="[run_until_cancel] cancelled_gracefully" + LOGS="" + for i in $(seq 1 30); do + LOGS="$(meshctl logs long-task-provider 2>&1)" + if echo "$LOGS" | grep -qF "$MARKER"; then + echo "$LOGS" | grep -F "$MARKER" + exit 0 + fi + sleep 1 + done + echo "ERROR: producer never logged the cancelled_gracefully marker" >&2 + echo "$LOGS" | tail -80 >&2 || true + exit 1 capture: producer_log + timeout: 60 # Independent verification: even if the producer didn't observe the # synthetic 'cancelled' event in its recv_event branch (e.g., because diff --git a/tests/integration/suites/uc22_meshjob_ts/tc26_cancel_posts_synthetic_event_ts/test.yaml b/tests/integration/suites/uc22_meshjob_ts/tc26_cancel_posts_synthetic_event_ts/test.yaml index 83dcc97ba..78e44120f 100644 --- a/tests/integration/suites/uc22_meshjob_ts/tc26_cancel_posts_synthetic_event_ts/test.yaml +++ b/tests/integration/suites/uc22_meshjob_ts/tc26_cancel_posts_synthetic_event_ts/test.yaml @@ -83,20 +83,31 @@ test: capture: commission_response timeout: 60 - # Allow a moment for the producer's stdout to flush after the - # cancelled event arrives. The producer's marker line is logged - # inside the recvEvent branch and the runtime captures stdout to - # the agent's log file. - - name: "Wait for producer log to flush" - handler: wait - seconds: 3 - - - name: "Read producer log for the cancelled_gracefully marker" + # Poll the producer's agent log until the cancelled_gracefully marker + # appears. The producer's marker line is logged inside the recvEvent + # branch and the runtime captures stdout to the agent's log file + # ASYNCHRONOUSLY — under CI load a fixed wait is sometimes too short + # (issue #1134). Poll up to ~30s, capturing the matching marker line + # as soon as it is present; on timeout dump logs and fail loud. + - name: "Poll producer log for the cancelled_gracefully marker" handler: shell workdir: /workspace command: | - meshctl logs long-task-provider-ts 2>&1 | tail -200 | grep "\[run_until_cancel\] cancelled_gracefully" || true + MARKER="[run_until_cancel] cancelled_gracefully" + LOGS="" + for i in $(seq 1 30); do + LOGS="$(meshctl logs long-task-provider-ts 2>&1)" + if echo "$LOGS" | grep -qF "$MARKER"; then + echo "$LOGS" | grep -F "$MARKER" + exit 0 + fi + sleep 1 + done + echo "ERROR: producer never logged the cancelled_gracefully marker" >&2 + echo "$LOGS" | tail -80 >&2 || true + exit 1 capture: producer_log + timeout: 60 # Independent verification: even if the producer didn't observe the # synthetic 'cancelled' event in its recvEvent branch (e.g., because diff --git a/tests/integration/suites/uc23_meshjob_java/tc26_cancel_posts_synthetic_event_java/test.yaml b/tests/integration/suites/uc23_meshjob_java/tc26_cancel_posts_synthetic_event_java/test.yaml index 258827c36..4246f434b 100644 --- a/tests/integration/suites/uc23_meshjob_java/tc26_cancel_posts_synthetic_event_java/test.yaml +++ b/tests/integration/suites/uc23_meshjob_java/tc26_cancel_posts_synthetic_event_java/test.yaml @@ -113,19 +113,31 @@ test: capture: commission_response timeout: 60 - # Allow a moment for the producer's stdout to flush after the cancel - # event arrives. The producer's marker line is printed inside the - # recvEvent loop and the runtime captures stdout to the agent's log. - - name: "Wait for producer log to flush" - handler: wait - seconds: 3 - - - name: "Read producer log for the cancelled_gracefully marker" + # Poll the producer's agent log until the cancelled_gracefully marker + # appears. The producer's marker line is printed inside the recvEvent + # loop and the runtime captures stdout to the agent's log + # ASYNCHRONOUSLY — under CI load a fixed wait is sometimes too short + # (issue #1134). Poll up to ~30s, capturing the matching marker line + # as soon as it is present; on timeout dump logs and fail loud. + - name: "Poll producer log for the cancelled_gracefully marker" handler: shell workdir: /workspace command: | - meshctl logs long-task-provider-java 2>&1 | tail -200 | grep "\[run_until_cancel\] cancelled_gracefully" || true + MARKER="[run_until_cancel] cancelled_gracefully" + LOGS="" + for i in $(seq 1 30); do + LOGS="$(meshctl logs long-task-provider-java 2>&1)" + if echo "$LOGS" | grep -qF "$MARKER"; then + echo "$LOGS" | grep -F "$MARKER" + exit 0 + fi + sleep 1 + done + echo "ERROR: producer never logged the cancelled_gracefully marker" >&2 + echo "$LOGS" | tail -80 >&2 || true + exit 1 capture: producer_log + timeout: 60 # Independent verification: even if the producer didn't observe the # synthetic 'cancelled' event in its recvEvent branch (e.g., because