diff --git a/pom.xml b/pom.xml index fda4114..5a2afec 100644 --- a/pom.xml +++ b/pom.xml @@ -101,6 +101,7 @@ org.postgresql postgresql + 42.7.11 org.flywaydb diff --git a/src/main/java/it/cnr/isti/workflow/manager/assistant/AssistantConversationService.java b/src/main/java/it/cnr/isti/workflow/manager/assistant/AssistantConversationService.java index 10862c6..66e1972 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/assistant/AssistantConversationService.java +++ b/src/main/java/it/cnr/isti/workflow/manager/assistant/AssistantConversationService.java @@ -1,5 +1,6 @@ package it.cnr.isti.workflow.manager.assistant; +import java.time.Duration; import java.time.Instant; import java.util.ArrayList; import java.util.List; @@ -9,6 +10,9 @@ import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import org.springframework.stereotype.Service; import org.springframework.web.server.ResponseStatusException; @@ -37,15 +41,20 @@ import jakarta.annotation.PreDestroy; public class AssistantConversationService { private static final int DEFAULT_MAX_REPAIR_ATTEMPTS = 1; + private static final Duration SESSION_TTL = Duration.ofHours(2); + private static final Duration CALL_TTL = Duration.ofHours(1); private final ConcurrentHashMap sessions = new ConcurrentHashMap<>(); private final ConcurrentHashMap calls = new ConcurrentHashMap<>(); private final ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor(); + private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor( + r -> { Thread t = new Thread(r, "assistant-cleanup"); t.setDaemon(true); return t; }); private final FlowAssistantService flowAssistantService; public AssistantConversationService(FlowAssistantService flowAssistantService) { this.flowAssistantService = flowAssistantService; + scheduler.scheduleAtFixedRate(this::evictExpired, 10, 10, TimeUnit.MINUTES); } public AssistantSessionView createSession(String owner, AssistantSessionCreateRequest request) { @@ -78,6 +87,10 @@ public class AssistantConversationService { public AssistantCallAcceptedResponse submitMessage(String sessionId, String owner, AssistantSessionMessageRequest request) { SessionState session = requireSession(sessionId, owner); + if (!session.running.compareAndSet(false, true)) { + throw new ResponseStatusException(HttpStatus.CONFLICT, + "A call is already in progress for this session"); + } session.appendMessage(AssistantMessageRole.USER, request.message(), null); Instant now = Instant.now(); @@ -93,6 +106,14 @@ public class AssistantConversationService { @PreDestroy void shutdown() { executor.shutdownNow(); + scheduler.shutdownNow(); + } + + private void evictExpired() { + Instant sessionCutoff = Instant.now().minus(SESSION_TTL); + Instant callCutoff = Instant.now().minus(CALL_TTL); + sessions.entrySet().removeIf(e -> e.getValue().updatedAt.isBefore(sessionCutoff)); + calls.entrySet().removeIf(e -> e.getValue().updatedAt.isBefore(callCutoff)); } private void processCall(SessionState session, CallState call, String message) { @@ -153,21 +174,45 @@ public class AssistantConversationService { "The assistant request failed: " + (e.getMessage() == null ? "unknown error" : e.getMessage()), call.id); session.touch(); + } finally { + session.running.set(false); } } private AssistantIntent inferIntent(SessionState session, String message) { String normalized = message.toLowerCase(Locale.ROOT); - if (normalized.contains("explain") || normalized.contains("what does") || normalized.contains("why")) { + + // FIX takes priority: explicit errors/flow-invalid state must not be swallowed by EXPLAIN + boolean flowInvalid = session.currentFlow != null + && session.lastValidationErrors != null + && !session.lastValidationErrors.isEmpty(); + if (flowInvalid + || normalized.contains("fix") + || normalized.contains("correggi") + || normalized.contains("ripara") + || normalized.contains("problem") + || normalized.contains("problema") + || normalized.contains("error") + || normalized.contains("errore")) { + if (session.currentFlow != null) { + return AssistantIntent.FIX; + } + } + + if (normalized.contains("explain") + || normalized.contains("spiega") + || normalized.contains("descrivi") + || normalized.contains("what does") + || normalized.contains("cosa fa") + || normalized.contains("why") + || normalized.contains("perché") + || normalized.contains("perche")) { return AssistantIntent.EXPLAIN; } + if (session.currentFlow == null) { return AssistantIntent.DRAFT; } - boolean flowInvalid = session.lastValidationErrors != null && !session.lastValidationErrors.isEmpty(); - if (flowInvalid || normalized.contains("fix") || normalized.contains("problem") || normalized.contains("error")) { - return AssistantIntent.FIX; - } return AssistantIntent.REFINE; } @@ -217,6 +262,7 @@ public class AssistantConversationService { private volatile String lastCallId; private volatile FlowCreateRequest currentFlow; private volatile List lastValidationErrors = List.of(); + private final AtomicBoolean running = new AtomicBoolean(false); private final List messages = java.util.Collections.synchronizedList(new ArrayList<>()); private SessionState(String id, String owner, String model, Instant createdAt, Instant updatedAt) { diff --git a/src/main/java/it/cnr/isti/workflow/manager/assistant/BlockCatalogService.java b/src/main/java/it/cnr/isti/workflow/manager/assistant/BlockCatalogService.java index 4d313a2..2f15601 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/assistant/BlockCatalogService.java +++ b/src/main/java/it/cnr/isti/workflow/manager/assistant/BlockCatalogService.java @@ -44,6 +44,7 @@ public class BlockCatalogService { String name, String type, boolean required, + List allowedValues, String placeholder, boolean structural) { } @@ -133,10 +134,12 @@ public class BlockCatalogService { Set requiredFields = extractRequiredFields(schema.path("required")); return schema.path("properties").properties().stream() .filter(entry -> !"type".equals(entry.getKey())) + .filter(entry -> !isSystemManagedField(entry.getKey())) .map(entry -> new AssistantPromptFieldDescriptor( entry.getKey(), extractFieldType(entry.getValue()), requiredFields.contains(entry.getKey()), + extractAllowedValues(entry.getValue(), schema), textOrNull(entry.getValue().path("x-ui-placeholder")), entry.getValue().path("x-ui-structural").asBoolean(false))) .sorted(Comparator @@ -145,6 +148,10 @@ public class BlockCatalogService { .toList(); } + private boolean isSystemManagedField(String fieldName) { + return Set.of("provider", "model", "llmDescriptor", "ids", "inputs", "outputs", "skills").contains(fieldName); + } + private List extractIoNames(Block exampleBlock, boolean inputs) { if (exampleBlock == null) { return List.of(); @@ -188,6 +195,42 @@ public class BlockCatalogService { return "object"; } + private List extractAllowedValues(JsonNode node, JsonNode schema) { + JsonNode enumNode = node.path("enum"); + if (enumNode instanceof ArrayNode enumValues) { + return extractStringValues(enumValues); + } + + String ref = textOrNull(node.path("$ref")); + if (ref == null) { + return List.of(); + } + + String definitionName = ref.startsWith("#/$defs/") ? ref.substring("#/$defs/".length()) + : ref.startsWith("#/definitions/") ? ref.substring("#/definitions/".length()) + : null; + if (definitionName == null) { + return List.of(); + } + + JsonNode referencedEnum = schema.path("$defs").path(definitionName).path("enum"); + if (referencedEnum.isMissingNode()) { + referencedEnum = schema.path("definitions").path(definitionName).path("enum"); + } + if (referencedEnum instanceof ArrayNode enumValues) { + return extractStringValues(enumValues); + } + return List.of(); + } + + private List extractStringValues(ArrayNode values) { + java.util.ArrayList result = new java.util.ArrayList<>(); + for (JsonNode value : values) { + value.stringValueOpt().ifPresent(result::add); + } + return result; + } + private String textOrNull(JsonNode node) { return node == null ? null : node.stringValueOpt().orElse(null); } diff --git a/src/main/java/it/cnr/isti/workflow/manager/assistant/FlowAssistantPromptService.java b/src/main/java/it/cnr/isti/workflow/manager/assistant/FlowAssistantPromptService.java index 722e7b0..d9fd526 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/assistant/FlowAssistantPromptService.java +++ b/src/main/java/it/cnr/isti/workflow/manager/assistant/FlowAssistantPromptService.java @@ -43,6 +43,8 @@ public class FlowAssistantPromptService { - Use only block types from the catalog. - Keep the plan minimal. - One block per logical action. + - LLMBlock is stateless across workflow nodes: use it for one-shot transformations where all required context is supplied as explicit inputs. + - When a later step must reuse memory, a session, an index, retrieved knowledge, RAG context, or any other shared working state produced by an earlier step, use MCPAgent blocks for both the producer and the consumer so they can share an MCP session. - blockId must be stable symbolic ids like b1, b2, b3. - Do not return block configuration yet. - Do not return connections yet. @@ -87,7 +89,11 @@ public class FlowAssistantPromptService { Rules: - Configure only the requested block. - The config object must contain only task-specific fields. - - Do not include system-managed fields like provider, model, llmDescriptor, ids, inputs, outputs. + - For enum fields, use only the allowedValues listed in the block descriptor. + - For HTTPServerCall without authentication, set requiresAuthorization to false and omit authorization details. + - For an MCPAgent block that creates shared memory, an index, reusable knowledge, or persistent state, set shareSession to true, sharedSessionName to "sharedMemorySession", useSharedSession to false, and choose the MCP server that matches the task. + - For an MCPAgent block that consumes shared memory, an index, reusable knowledge, or persistent state, set useSharedSession to true, sharedSessionRef to "sharedMemorySession", shareSession to false, and include both an upstream ordering placeholder such as ${{state_ready}} and any user input placeholder such as ${{query}} in the prompt. + - Do not include system-managed fields like provider, model, llmDescriptor, ids, inputs, outputs, skills. - Use placeholders like ${{variable}} when needed. - Return valid JSON with no markdown fences. @@ -147,8 +153,11 @@ public class FlowAssistantPromptService { - Never use block names or purposes in fromBlockId/toBlockId. - Use only output/input names that exist in the configured blocks. - Prefer canonical names such as response, output, input, and prompt when they exist. + - Do not leave processing blocks disconnected when one step depends on another step's output or execution order. + - For MCPAgent blocks that reuse a shared session, still connect the producer response to the consumer's upstream ordering placeholder, for example state_ready, so execution order is explicit. - Allowed block ids are listed in the flow plan and configured blocks. Use only those exact values. - Return the minimal set of connections required by the user request. + - If no connections are needed, still return JSON with "connections": []. - Return valid JSON with no markdown fences. Flow plan: @@ -189,6 +198,25 @@ public class FlowAssistantPromptService { """.formatted(toJson(flow), userPrompt == null || userPrompt.isBlank() ? "general explanation" : userPrompt); } + public String buildStructuredRepairPrompt(String originalPrompt, String invalidResponse, String errorMessage) { + return """ + The previous response did not satisfy the required JSON contract. + Return only corrected valid JSON for the same task. Do not include markdown fences or explanatory text. + + Error: + %s + + Previous invalid response: + %s + + Original task and contract: + %s + """.formatted( + truncate(errorMessage, 2000), + truncate(invalidResponse, 4000), + originalPrompt); + } + private String toJson(Object value) { try { return ObjectMapperHolder.mapper.writerWithDefaultPrettyPrinter().writeValueAsString(value); @@ -210,4 +238,14 @@ public class FlowAssistantPromptService { } return toJson(errors); } + + private String truncate(String value, int maxLength) { + if (value == null) { + return "(none)"; + } + if (value.length() <= maxLength) { + return value; + } + return value.substring(0, maxLength) + "\n...(truncated)"; + } } diff --git a/src/main/java/it/cnr/isti/workflow/manager/assistant/FlowAssistantService.java b/src/main/java/it/cnr/isti/workflow/manager/assistant/FlowAssistantService.java index 0b1fe03..65a3078 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/assistant/FlowAssistantService.java +++ b/src/main/java/it/cnr/isti/workflow/manager/assistant/FlowAssistantService.java @@ -13,9 +13,14 @@ import java.util.Set; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.http.HttpStatus; import org.springframework.stereotype.Service; +import org.springframework.beans.factory.annotation.Value; import org.springframework.web.server.ResponseStatusException; +import tools.jackson.core.json.JsonReadFeature; import tools.jackson.databind.JsonNode; +import tools.jackson.databind.ObjectMapper; +import tools.jackson.databind.json.JsonMapper; +import tools.jackson.databind.node.ArrayNode; import tools.jackson.databind.node.ObjectNode; import it.cnr.isti.workflow.manager.app.ObjectMapperHolder; @@ -28,6 +33,7 @@ import it.cnr.isti.workflow.manager.assistant.model.AssistantGenerationRequest; import it.cnr.isti.workflow.manager.assistant.model.AssistantRefineRequest; import it.cnr.isti.workflow.manager.blocks.Block; import it.cnr.isti.workflow.manager.blocks.configurations.BlockConfiguration; +import it.cnr.isti.workflow.manager.blocks.configurations.MCPAgentBlockConfiguration; import it.cnr.isti.workflow.manager.blocks.factories.BlockFactory; import it.cnr.isti.workflow.manager.flows.model.Connection; import it.cnr.isti.workflow.manager.flows.model.FlowCreateRequest; @@ -43,10 +49,26 @@ import jakarta.validation.Validator; public class FlowAssistantService { private static final String INTERNAL_PROVIDER_NAME = "InternalOllama"; + private static final String SHARED_MEMORY_SESSION_NAME = "sharedMemorySession"; + private static final int DEFAULT_PROVIDER_RETRY_ATTEMPTS = 3; + private static final long DEFAULT_RETRY_BASE_DELAY_MILLIS = 120L; + private static final long DEFAULT_RETRY_MAX_DELAY_MILLIS = 800L; + private static final ObjectMapper LENIENT_ASSISTANT_MAPPER = JsonMapper.builder() + .enable(JsonReadFeature.ALLOW_UNQUOTED_PROPERTY_NAMES) + .enable(JsonReadFeature.ALLOW_SINGLE_QUOTES) + .enable(JsonReadFeature.ALLOW_TRAILING_COMMA) + .build(); + @FunctionalInterface public interface ProgressListener { void onProgress(String phase, String message); } + + @FunctionalInterface + private interface StructuredResponseParser { + T parse(String rawResponse); + } + private static final ProgressListener NOOP_PROGRESS = (phase, message) -> { }; @@ -68,6 +90,9 @@ public class FlowAssistantService { private record ParsedBlockDraft(AssistantConfiguredBlockDraft block, String rationale) { } + private record ConfiguredBlockResult(ParsedBlockDraft parsedBlock, Block block) { + } + private record ParsedConnections(List connections, String rationale) { } @@ -93,6 +118,15 @@ public class FlowAssistantService { @Autowired private Validator validator; + @Value("${app.assistant.provider-retry-base-delay-ms:" + DEFAULT_RETRY_BASE_DELAY_MILLIS + "}") + private long retryBaseDelayMillis; + + @Value("${app.assistant.provider-retry-max-delay-ms:" + DEFAULT_RETRY_MAX_DELAY_MILLIS + "}") + private long retryMaxDelayMillis; + + @Value("${app.assistant.provider-retry-attempts:" + DEFAULT_PROVIDER_RETRY_ATTEMPTS + "}") + private int providerRetryAttempts; + public AssistantFlowResponse draft(AssistantGenerationRequest request) { return draft(request, NOOP_PROGRESS); } @@ -180,9 +214,13 @@ public class FlowAssistantService { } progressListener.onProgress("planning", "Planning workflow blocks"); - ParsedPlan parsedPlan = parsePlan(invokeStructuredProvider(provider, model, - promptService.buildPlanPrompt(mode, userPrompt, currentFlow, errors, catalog))); - validatePlan(parsedPlan.plan()); + String planPrompt = promptService.buildPlanPrompt(mode, userPrompt, currentFlow, errors, catalog); + ParsedPlan parsedPlan = invokeStructuredAndValidate(provider, model, planPrompt, "plan", rawResponse -> { + ParsedPlan plan = parsePlan(rawResponse); + validatePlan(plan.plan(), userPrompt, currentFlow, catalogByType.keySet()); + return plan; + }); + boolean requireSharedMemorySemantics = isSharedMemoryContext(userPrompt, currentFlow, parsedPlan.plan()); List> assembledBlocks = new ArrayList<>(); Map> blocksByPlanId = new LinkedHashMap<>(); @@ -199,13 +237,20 @@ public class FlowAssistantService { "Assistant selected an unknown block type: " + blockPlan.blockType()); } - ParsedBlockDraft parsedBlock = parseBlockDraft(invokeStructuredProvider(provider, model, - promptService.buildBlockConfigurationPrompt(mode, userPrompt, descriptor, parsedPlan.plan(), blockPlan, - currentFlow, errors, model))); - appendRationale(rationaleParts, parsedBlock.rationale()); + String blockPrompt = promptService.buildBlockConfigurationPrompt(mode, userPrompt, descriptor, + parsedPlan.plan(), blockPlan, currentFlow, errors, model); + ConfiguredBlockResult configuredBlock = invokeStructuredAndValidate(provider, model, blockPrompt, + "block configuration for " + blockPlan.blockId(), rawResponse -> { + ParsedBlockDraft parsedBlock = parseBlockDraft(rawResponse); + AssistantConfiguredBlockDraft normalizedDraft = normalizeBlockDraft(blockPlan, + parsedBlock.block()); + Block assembledBlock = buildBlock(descriptor, blockPlan, normalizedDraft, model, + requireSharedMemorySemantics); + return new ConfiguredBlockResult(parsedBlock, assembledBlock); + }); + appendRationale(rationaleParts, configuredBlock.parsedBlock().rationale()); - AssistantConfiguredBlockDraft normalizedDraft = normalizeBlockDraft(blockPlan, parsedBlock.block()); - Block assembledBlock = buildBlock(descriptor, blockPlan, normalizedDraft, model); + Block assembledBlock = configuredBlock.block(); assembledBlocks.add(assembledBlock); blocksByPlanId.put(blockPlan.blockId(), assembledBlock); registerBlockAlias(blocksByAlias, blockPlan.blockId(), assembledBlock); @@ -223,9 +268,20 @@ public class FlowAssistantService { } progressListener.onProgress("connecting_blocks", "Connecting configured blocks"); - ParsedConnections parsedConnections = parseConnections(invokeStructuredProvider(provider, model, - promptService.buildConnectionsPrompt(mode, userPrompt, parsedPlan.plan(), configuredBlocks, currentFlow, - errors))); + String connectionsPrompt = promptService.buildConnectionsPrompt(mode, userPrompt, parsedPlan.plan(), + configuredBlocks, currentFlow, errors); + ParsedConnections parsedConnections = invokeStructuredAndValidate(provider, model, connectionsPrompt, + "connections", rawResponse -> { + ParsedConnections parsed = parseConnectionsOrInferSequential(rawResponse, assembledBlocks); + parsed = completeRequiredSequentialConnections(requireSharedMemorySemantics, parsed, + assembledBlocks, blocksByPlanId, blocksByAlias); + List candidateConnections = parsed.connections().stream() + .map(connection -> toConnection(connection, blocksByPlanId, blocksByAlias)) + .toList(); + validateSharedMemorySemantics(requireSharedMemorySemantics, blocksByPlanId.values(), + candidateConnections); + return parsed; + }); appendRationale(rationaleParts, parsedConnections.rationale()); List connections = parsedConnections.connections().stream() @@ -243,6 +299,93 @@ public class FlowAssistantService { return new AssembledFlow(flow, String.join(" ", rationaleParts).trim()); } + private ParsedConnections parseConnectionsOrInferSequential(String rawResponse, List> assembledBlocks) { + try { + return parseConnections(rawResponse); + } catch (ResponseStatusException e) { + if (e.getReason() == null || !e.getReason().contains("invalid connections payload")) { + throw e; + } + List inferred = inferSequentialConnections(assembledBlocks); + if (inferred.isEmpty()) { + throw e; + } + return new ParsedConnections(inferred, + rawResponse == null || rawResponse.isBlank() + ? "Inferred sequential block connections." + : rawResponse.trim()); + } + } + + private T invokeStructuredAndValidate(LLMProvider provider, String model, String prompt, String taskName, + StructuredResponseParser parser) { + String currentPrompt = prompt; + String rawResponse = null; + ResponseStatusException lastFailure = null; + int maxAttempts = maxProviderRetryAttempts(); + + for (int attempt = 1; attempt <= maxAttempts; attempt++) { + try { + rawResponse = invokeStructuredProvider(provider, model, currentPrompt); + } catch (ResponseStatusException e) { + lastFailure = e; + if (attempt < maxAttempts && isRetriableProviderFailure(e)) { + waitBeforeRetry(attempt); + continue; + } + throw e; + } + try { + return parser.parse(rawResponse); + } catch (ResponseStatusException e) { + lastFailure = e; + } catch (RuntimeException e) { + lastFailure = new ResponseStatusException(HttpStatus.BAD_GATEWAY, + "Assistant returned an invalid " + taskName + " payload: " + e.getMessage(), e); + } + + if (attempt < maxAttempts) { + currentPrompt = promptService.buildStructuredRepairPrompt( + prompt, + rawResponse, + lastFailure == null ? "unknown validation error" : lastFailure.getReason()); + } + } + + throw lastFailure == null + ? new ResponseStatusException(HttpStatus.BAD_GATEWAY, + "Assistant returned an invalid " + taskName + " payload") + : lastFailure; + } + + private int maxProviderRetryAttempts() { + return Math.max(1, providerRetryAttempts); + } + + private boolean isRetriableProviderFailure(ResponseStatusException error) { + if (error == null || error.getStatusCode() == null) { + return false; + } + int statusCode = error.getStatusCode().value(); + return statusCode >= 500 && statusCode < 600; + } + + private void waitBeforeRetry(int attempt) { + long baseDelay = Math.max(0L, retryBaseDelayMillis); + long maxDelay = Math.max(baseDelay, retryMaxDelayMillis); + if (baseDelay == 0L || maxDelay == 0L) { + return; + } + long delay = Math.min(baseDelay * (1L << Math.max(0, attempt - 1)), maxDelay); + try { + Thread.sleep(delay); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new ResponseStatusException(HttpStatus.BAD_GATEWAY, + "Assistant retry interrupted", e); + } + } + private AssistantConfiguredBlockDraft normalizeBlockDraft(AssistantBlockPlan blockPlan, AssistantConfiguredBlockDraft draft) { if (draft == null) { @@ -260,7 +403,7 @@ public class FlowAssistantService { } private Block buildBlock(BlockCatalogService.AssistantPromptBlockDescriptor descriptor, AssistantBlockPlan blockPlan, - AssistantConfiguredBlockDraft draft, String model) { + AssistantConfiguredBlockDraft draft, String model, boolean requireSharedMemorySemantics) { if (!(draft.config() instanceof ObjectNode configNode)) { throw new ResponseStatusException(HttpStatus.BAD_GATEWAY, "Assistant returned a non-object config for block " + blockPlan.blockId()); @@ -270,6 +413,8 @@ public class FlowAssistantService { normalizedConfig.put("type", descriptor.configurationType()); normalizedConfig.put("name", defaultIfBlank(draft.name(), defaultIfBlank(blockPlan.purpose(), blockPlan.blockType()))); injectSystemManagedFields(normalizedConfig, descriptor, model); + normalizeHttpServerCallAuthorization(normalizedConfig, blockPlan); + normalizeMcpAgentSharedMemory(normalizedConfig, blockPlan, model, requireSharedMemorySemantics); try { BlockConfiguration configuration = ObjectMapperHolder.mapper.treeToValue(normalizedConfig, @@ -282,32 +427,119 @@ public class FlowAssistantService { } } - private void injectSystemManagedFields(ObjectNode config, BlockCatalogService.AssistantPromptBlockDescriptor descriptor, - String model) { - switch (descriptor.configurationType()) { - case "LLMBlockConfiguration" -> config.set("llmDescriptor", llmDescriptorNode(model)); - case "ChatInteractionBlockConfiguration" -> config.set("llmDescriptor", llmDescriptorNode(model)); - case "ConditionalBlockConfiguration" -> { - boolean useLlm = inferConditionalUseLlm(config); - config.put("useLlm", useLlm); - if (useLlm) { - config.set("llmDescriptor", llmDescriptorNode(model)); - } else { - config.remove("llmDescriptor"); - } + private void normalizeHttpServerCallAuthorization(ObjectNode config, AssistantBlockPlan blockPlan) { + if (!"HTTPServerCall".equals(blockPlan.blockType())) { + return; + } + + JsonNode requiresAuthorizationNode = config.get("requiresAuthorization"); + boolean requiresAuthorization = requiresAuthorizationNode != null + && !requiresAuthorizationNode.isNull() + && requiresAuthorizationNode.asBoolean(false); + + JsonNode authorizationTypeNode = config.get("authorizationType"); + String authorizationType = authorizationTypeNode == null || authorizationTypeNode.isNull() + ? null + : authorizationTypeNode.stringValueOpt().orElse(null); + + if (authorizationType == null || authorizationType.isBlank()) { + if (!requiresAuthorization) { + config.put("requiresAuthorization", false); + config.put("authorizationType", "API_KEY"); } - case "SwitchBlockConfiguration" -> { - boolean useLlm = inferConditionalUseLlm(config); - config.put("useLlm", useLlm); - if (useLlm) { - config.set("llmDescriptor", llmDescriptorNode(model)); - } else { - config.remove("llmDescriptor"); - } + return; + } + + if ("NONE".equalsIgnoreCase(authorizationType)) { + config.put("requiresAuthorization", false); + config.put("authorizationType", "API_KEY"); + } + } + + private void normalizeMcpAgentSharedMemory(ObjectNode config, AssistantBlockPlan blockPlan, String model, + boolean requireSharedMemorySemantics) { + if (!requireSharedMemorySemantics || !"MCPAgent".equals(blockPlan.blockType())) { + return; + } + + if (isSharedStateConsumerPurpose(blockPlan.purpose())) { + config.put("shareSession", false); + config.put("useSharedSession", true); + if (!hasTextValue(config.get("sharedSessionRef"))) { + config.put("sharedSessionRef", SHARED_MEMORY_SESSION_NAME); } - default -> { + return; + } + + if (isSharedStateProducerPurpose(blockPlan.purpose())) { + config.put("shareSession", true); + config.put("useSharedSession", false); + if (!hasTextValue(config.get("sharedSessionName"))) { + config.put("sharedSessionName", SHARED_MEMORY_SESSION_NAME); + } + if (!hasTextValue(config.get("model"))) { + config.put("model", model); + } + ensureDefaultRagServer(config); + return; + } + } + + private void ensureDefaultRagServer(ObjectNode config) { + if (hasMcpServer(config, "rag")) { + return; + } + JsonNode existingServers = config.get("mcpServers"); + if (existingServers != null && existingServers.isArray() && !existingServers.isEmpty()) { + return; + } + ArrayNode servers = config.putArray("mcpServers"); + ObjectNode ragServer = ObjectMapperHolder.mapper.createObjectNode(); + ragServer.put("sourceType", "CATALOG"); + ragServer.put("serverName", "rag"); + ragServer.set("configuration", ObjectMapperHolder.mapper.createObjectNode()); + servers.add(ragServer); + } + + private boolean hasMcpServer(ObjectNode config, String serverName) { + JsonNode servers = config.get("mcpServers"); + if (servers == null || !servers.isArray()) { + return false; + } + for (JsonNode server : servers) { + if (serverName.equalsIgnoreCase(textOrEmpty(server.path("serverName")))) { + return true; } } + return false; + } + + private void injectSystemManagedFields(ObjectNode config, BlockCatalogService.AssistantPromptBlockDescriptor descriptor, + String model) { + String configurationType = descriptor.configurationType(); + removeSystemManagedFields(config); + if (isConfigurationType(configurationType, "LLMBlockConfiguration") + || isConfigurationType(configurationType, "ChatInteractionBlockConfiguration")) { + config.set("llmDescriptor", llmDescriptorNode(model)); + return; + } + if (isConfigurationType(configurationType, "ConditionalBlockConfiguration") + || isConfigurationType(configurationType, "SwitchBlockConfiguration")) { + boolean useLlm = inferConditionalUseLlm(config); + config.put("useLlm", useLlm); + if (useLlm) { + config.set("llmDescriptor", llmDescriptorNode(model)); + } + } + } + + private void removeSystemManagedFields(ObjectNode config) { + config.remove(List.of("provider", "model", "llmDescriptor", "ids", "inputs", "outputs", "skills")); + } + + private boolean isConfigurationType(String actualType, String expectedSimpleName) { + return actualType != null + && (actualType.equals(expectedSimpleName) || actualType.endsWith("." + expectedSimpleName)); } private boolean inferConditionalUseLlm(ObjectNode config) { @@ -346,9 +578,19 @@ public class FlowAssistantService { } String sourceName = resolveConnectionOutputName(source, connection.fromOutput()); String targetName = resolveConnectionInputName(target, connection.toInput()); - if (sourceName == null || targetName == null) { + if (sourceName == null) { throw new ResponseStatusException(HttpStatus.BAD_GATEWAY, - "Assistant returned a connection with unresolved input/output names"); + "Assistant returned a connection with unresolved output '" + connection.fromOutput() + + "' on block '" + source.getName() + "' (available: " + + (source.getOutputs() == null ? "none" : source.getOutputs().stream() + .map(io -> io.getName()).toList()) + ")"); + } + if (targetName == null) { + throw new ResponseStatusException(HttpStatus.BAD_GATEWAY, + "Assistant returned a connection with unresolved input '" + connection.toInput() + + "' on block '" + target.getName() + "' (available: " + + (target.getInputs() == null ? "none" : target.getInputs().stream() + .map(io -> io.getName()).toList()) + ")"); } return Connection.builder() .sourceId(source.getId()) @@ -358,6 +600,86 @@ public class FlowAssistantService { .build(); } + private List inferSequentialConnections(List> blocks) { + if (blocks == null || blocks.size() < 2) { + return List.of(); + } + + List connections = new ArrayList<>(); + for (int i = 0; i < blocks.size() - 1; i++) { + Block source = blocks.get(i); + Block target = blocks.get(i + 1); + String sourceOutput = resolveConnectionOutputName(source, "response"); + String targetInput = resolveSequentialInputName(target); + if (sourceOutput == null || targetInput == null) { + continue; + } + connections.add(new AssistantConnectionDraft( + source.getName(), + sourceOutput, + target.getName(), + targetInput)); + } + return connections; + } + + private ParsedConnections completeRequiredSequentialConnections(boolean required, ParsedConnections parsed, + List> assembledBlocks, Map> blocksByPlanId, Map> blocksByAlias) { + if (!required) { + return parsed; + } + + List merged = new ArrayList<>( + parsed == null || parsed.connections() == null ? List.of() : parsed.connections()); + for (AssistantConnectionDraft inferred : inferSequentialConnections(assembledBlocks)) { + Block inferredSource = resolveConnectionBlock(inferred.fromBlockId(), blocksByPlanId, blocksByAlias); + Block inferredTarget = resolveConnectionBlock(inferred.toBlockId(), blocksByPlanId, blocksByAlias); + if (inferredSource == null || inferredTarget == null) { + continue; + } + boolean alreadyConnected = false; + for (AssistantConnectionDraft existing : merged) { + Block existingSource = resolveConnectionBlock(existing.fromBlockId(), blocksByPlanId, blocksByAlias); + Block existingTarget = resolveConnectionBlock(existing.toBlockId(), blocksByPlanId, blocksByAlias); + if (existingSource != null && existingTarget != null + && Objects.equals(existingSource.getId(), inferredSource.getId()) + && Objects.equals(existingTarget.getId(), inferredTarget.getId())) { + alreadyConnected = true; + break; + } + } + if (!alreadyConnected) { + merged.add(inferred); + } + } + + String rationale = parsed == null ? "" : parsed.rationale(); + return new ParsedConnections(merged, + defaultIfBlank(rationale, "Completed required sequential shared-memory connections.")); + } + + private String resolveSequentialInputName(Block block) { + List inputs = block.getInputs(); + if (inputs == null || inputs.isEmpty()) { + return null; + } + if (inputs.size() == 1) { + return inputs.getFirst().getName(); + } + return inputs.stream() + .map(IODescriptor::getName) + .filter(name -> !isLikelyUserProvidedInput(name)) + .findFirst() + .orElse(inputs.getFirst().getName()); + } + + private boolean isLikelyUserProvidedInput(String name) { + String normalized = normalizeBlockReference(name); + return normalized != null + && Set.of("query", "question", "user_query", "userquery", "url", "file_url", "fileurl") + .contains(normalized); + } + private Block resolveConnectionBlock(String rawReference, Map> blocksByPlanId, Map> blocksByAlias) { if (rawReference == null || rawReference.isBlank()) { @@ -450,9 +772,23 @@ public class FlowAssistantService { return normalized.isBlank() ? null : normalized; } + private static boolean containsWord(String text, String word) { + int idx = text.indexOf(word); + while (idx >= 0) { + boolean boundaryBefore = idx == 0 || !Character.isLetterOrDigit(text.charAt(idx - 1)); + boolean boundaryAfter = idx + word.length() >= text.length() + || !Character.isLetterOrDigit(text.charAt(idx + word.length())); + if (boundaryBefore && boundaryAfter) { + return true; + } + idx = text.indexOf(word, idx + 1); + } + return false; + } + private ParsedPlan parsePlan(String rawResponse) { try { - JsonNode root = ObjectMapperHolder.mapper.readTree(extractJsonObject(rawResponse)); + JsonNode root = readJsonObject(rawResponse); JsonNode planNode = root.has("plan") ? root.get("plan") : root; AssistantFlowPlan plan = ObjectMapperHolder.mapper.treeToValue(planNode, AssistantFlowPlan.class); return new ParsedPlan(plan, textOrEmpty(root.path("rationale"))); @@ -464,7 +800,7 @@ public class FlowAssistantService { private ParsedBlockDraft parseBlockDraft(String rawResponse) { try { - JsonNode root = ObjectMapperHolder.mapper.readTree(extractJsonObject(rawResponse)); + JsonNode root = readJsonObject(rawResponse); JsonNode blockNode = root.has("block") ? root.get("block") : root; AssistantConfiguredBlockDraft block = new AssistantConfiguredBlockDraft( textOrNull(blockNode.path("blockId")), @@ -482,7 +818,7 @@ public class FlowAssistantService { private ParsedConnections parseConnections(String rawResponse) { try { - JsonNode root = ObjectMapperHolder.mapper.readTree(extractJsonObjectOrArray(rawResponse)); + JsonNode root = readJsonObjectOrArray(rawResponse); if (root.isArray()) { List arrayConnections = new ArrayList<>(); for (JsonNode node : root) { @@ -510,7 +846,34 @@ public class FlowAssistantService { } } - private void validatePlan(AssistantFlowPlan plan) { + private JsonNode readJsonObject(String rawResponse) { + String json = extractJsonObject(rawResponse); + try { + return ObjectMapperHolder.mapper.readTree(json); + } catch (Exception e) { + try { + return LENIENT_ASSISTANT_MAPPER.readTree(json); + } catch (Exception ignored) { + throw new IllegalArgumentException(e.getMessage(), e); + } + } + } + + private JsonNode readJsonObjectOrArray(String rawResponse) { + String json = extractJsonObjectOrArray(rawResponse); + try { + return ObjectMapperHolder.mapper.readTree(json); + } catch (Exception e) { + try { + return LENIENT_ASSISTANT_MAPPER.readTree(json); + } catch (Exception ignored) { + throw new IllegalArgumentException(e.getMessage(), e); + } + } + } + + private void validatePlan(AssistantFlowPlan plan, String userPrompt, FlowCreateRequest currentFlow, + Set availableBlockTypes) { if (plan == null) { throw new ResponseStatusException(HttpStatus.BAD_GATEWAY, "Assistant returned an empty plan"); } @@ -527,9 +890,215 @@ public class FlowAssistantService { throw new ResponseStatusException(HttpStatus.BAD_GATEWAY, "Assistant returned a block without blockType"); } + if (isSharedMemoryContext(userPrompt, currentFlow, plan) + && "LLMBlock".equals(block.blockType()) + && isSharedStatePurpose(block.purpose())) { + if (!availableBlockTypes.contains("MCPAgent")) { + // MCPAgent not available in this deployment: skip the constraint rather than + // silently accepting an invalid plan — surface a clear error so the operator + // knows the catalog is incomplete for shared-memory workflows. + throw new ResponseStatusException(HttpStatus.BAD_GATEWAY, + "Assistant returned a plan that requires shared-memory semantics but MCPAgent is not available in the block catalog"); + } + throw new ResponseStatusException(HttpStatus.BAD_GATEWAY, + "Assistant returned an inconsistent plan: workflows that require shared memory or reusable state between steps must use MCPAgent blocks with a shared MCP session instead of standalone LLMBlock nodes"); + } } } + private void validateSharedMemorySemantics(boolean required, Collection> blocks, + List connections) { + if (!required) { + return; + } + + Block sessionProducer = null; + Block sessionConsumer = null; + for (Block block : blocks) { + if (block.getSpecificConfiguration() instanceof MCPAgentBlockConfiguration configuration) { + if (Boolean.TRUE.equals(configuration.getShareSession()) + && SHARED_MEMORY_SESSION_NAME.equals(configuration.getSharedSessionName())) { + sessionProducer = block; + } + if (Boolean.TRUE.equals(configuration.getUseSharedSession()) + && SHARED_MEMORY_SESSION_NAME.equals(configuration.getSharedSessionRef())) { + sessionConsumer = block; + } + } + } + + if (sessionProducer == null || sessionConsumer == null) { + throw new ResponseStatusException(HttpStatus.BAD_GATEWAY, + "Assistant returned an inconsistent flow: shared-memory workflows require an MCPAgent producer and consumer sharing " + + SHARED_MEMORY_SESSION_NAME + ". Blocks: " + summarizeMcpSessionBlocks(blocks)); + } + if (!isReachable(sessionProducer.getId(), sessionConsumer.getId(), connections, new LinkedHashSet<>())) { + throw new ResponseStatusException(HttpStatus.BAD_GATEWAY, + "Assistant returned an inconsistent connections payload: the MCPAgent that consumes shared memory must execute after the producer MCPAgent"); + } + } + + private boolean isReachable(String sourceId, String targetId, List connections, Set visited) { + if (!visited.add(sourceId)) { + return false; + } + for (Connection connection : connections) { + if (!Objects.equals(sourceId, connection.getSourceId())) { + continue; + } + if (Objects.equals(targetId, connection.getTargetId()) + || isReachable(connection.getTargetId(), targetId, connections, visited)) { + return true; + } + } + return false; + } + + private String summarizeMcpSessionBlocks(Collection> blocks) { + if (blocks == null || blocks.isEmpty()) { + return "(none)"; + } + return blocks.stream() + .map(block -> { + if (block.getSpecificConfiguration() instanceof MCPAgentBlockConfiguration configuration) { + return block.getName() + + "[MCPAgent shareSession=" + configuration.getShareSession() + + ", sharedSessionName=" + configuration.getSharedSessionName() + + ", useSharedSession=" + configuration.getUseSharedSession() + + ", sharedSessionRef=" + configuration.getSharedSessionRef() + "]"; + } + return block.getName() + "[" + (block.getType() == null ? "unknown" : block.getType().getName()) + "]"; + }) + .toList() + .toString(); + } + + private boolean isSharedMemoryRequest(String text) { + String normalized = normalizeBlockReference(text); + if (normalized == null) { + return false; + } + boolean explicitMemory = normalized.contains("memory") + || normalized.contains("memoria") + || normalized.contains("session") + || normalized.contains("shared context") + || normalized.contains("contesto condiviso") + || normalized.contains("remember") + || normalized.contains("ricord") + || normalized.contains("stateful") + || normalized.contains("persistent") + || normalized.contains("persist") + || containsWord(normalized, "rag") + || normalized.contains("knowledge base") + || normalized.contains("base di conoscenza") + || normalized.contains("vector") + || normalized.contains("embedding"); + boolean producer = containsSharedStateProducerTerm(normalized); + boolean consumer = containsSharedStateConsumerTerm(normalized); + boolean crossStepMemory = explicitMemory + && (normalized.contains("share") + || normalized.contains("condiv") + || normalized.contains("reuse") + || normalized.contains("rius") + || normalized.contains("between") + || normalized.contains("across") + || normalized.contains("later") + || normalized.contains("subsequent") + || normalized.contains("node") + || normalized.contains("nodi") + || normalized.contains("step") + || normalized.contains("passi")); + return producer && consumer || crossStepMemory && producer; + } + + private boolean isSharedMemoryContext(String userPrompt, FlowCreateRequest currentFlow, AssistantFlowPlan plan) { + if (isSharedMemoryRequest(userPrompt)) { + return true; + } + if (plan != null) { + StringBuilder planText = new StringBuilder(); + planText.append(defaultIfBlank(plan.name(), "")).append(' ') + .append(defaultIfBlank(plan.description(), "")); + if (plan.blocks() != null) { + for (AssistantBlockPlan block : plan.blocks()) { + planText.append(' ').append(defaultIfBlank(block.purpose(), "")) + .append(' ').append(defaultIfBlank(block.blockType(), "")); + } + } + if (isSharedMemoryRequest(planText.toString())) { + return true; + } + } + if (currentFlow == null || currentFlow.flow() == null) { + return false; + } + StringBuilder flowText = new StringBuilder(); + flowText.append(defaultIfBlank(currentFlow.name(), "")).append(' ') + .append(defaultIfBlank(currentFlow.description(), "")); + if (currentFlow.flow().getBlocks() != null) { + for (Block block : currentFlow.flow().getBlocks()) { + flowText.append(' ') + .append(defaultIfBlank(block.getName(), "")) + .append(' ') + .append(block.getType() == null ? "" : defaultIfBlank(block.getType().getName(), "")); + } + } + return isSharedMemoryRequest(flowText.toString()); + } + + private boolean isSharedStatePurpose(String purpose) { + return isSharedStateProducerPurpose(purpose) || isSharedStateConsumerPurpose(purpose); + } + + private boolean isSharedStateProducerPurpose(String purpose) { + String normalized = normalizeBlockReference(purpose); + return normalized != null && containsSharedStateProducerTerm(normalized); + } + + private boolean isSharedStateConsumerPurpose(String purpose) { + String normalized = normalizeBlockReference(purpose); + return normalized != null && containsSharedStateConsumerTerm(normalized); + } + + private boolean containsSharedStateProducerTerm(String normalized) { + return normalized.contains("index") + || normalized.contains("indicizz") + || normalized.contains("ingest") + || containsWord(normalized, "store") + || normalized.contains("save") + || normalized.contains("archive") + || normalized.contains("remember") + || normalized.contains("memor") + || normalized.contains("knowledge") + || normalized.contains("conoscenza") + || normalized.contains("embedding") + || normalized.contains("vector") + || containsWord(normalized, "rag") + || normalized.contains("retrieve corpus") + || normalized.contains("retrieval corpus"); + } + + private boolean containsSharedStateConsumerTerm(String normalized) { + return normalized.contains("query") + || normalized.contains("queri") + || normalized.contains("answer") + || normalized.contains("question") + || normalized.contains("search") + || normalized.contains("retrieve") + || normalized.contains("retrieval") + || containsWord(normalized, "ask") + || normalized.contains("domand") + || normalized.contains("interrog") + || normalized.contains("consult") + || normalized.contains("use the index") + || normalized.contains("using indexed") + || normalized.contains("grounded"); + } + + private boolean hasTextValue(JsonNode node) { + return node != null && !node.isNull() && textOrNull(node) != null && !textOrNull(node).isBlank(); + } + private List validate(FlowCreateRequest flow) { Set> violations = validator.validate(flow); if (violations.isEmpty()) { @@ -564,38 +1133,19 @@ public class FlowAssistantService { } String trimmed = rawResponse.trim(); - int start = trimmed.indexOf('{'); - if (start < 0) { - throw new IllegalArgumentException("No JSON object found in assistant response"); + boolean sawObjectStart = false; + for (int i = 0; i < trimmed.length(); i++) { + if (trimmed.charAt(i) != '{') { + continue; + } + sawObjectStart = true; + String candidate = tryExtractBalancedJson(trimmed, i); + if (candidate != null) { + return candidate; + } } - int depth = 0; - boolean inString = false; - boolean escaped = false; - for (int i = start; i < trimmed.length(); i++) { - char current = trimmed.charAt(i); - if (escaped) { - escaped = false; - continue; - } - if (current == '\\' && inString) { - escaped = true; - continue; - } - if (current == '"') { - inString = !inString; - continue; - } - if (inString) { - continue; - } - if (current == '{') { - depth++; - } else if (current == '}') { - depth--; - if (depth == 0) { - return trimmed.substring(start, i + 1); - } - } + if (!sawObjectStart) { + throw new IllegalArgumentException("No JSON object found in assistant response"); } throw new IllegalArgumentException("Incomplete JSON object found in assistant response"); } @@ -606,26 +1156,31 @@ public class FlowAssistantService { } String trimmed = rawResponse.trim(); - int objectStart = trimmed.indexOf('{'); - int arrayStart = trimmed.indexOf('['); - int start = -1; - if (objectStart >= 0 && arrayStart >= 0) { - start = Math.min(objectStart, arrayStart); - } else if (objectStart >= 0) { - start = objectStart; - } else if (arrayStart >= 0) { - start = arrayStart; + boolean sawJsonStart = false; + for (int i = 0; i < trimmed.length(); i++) { + char current = trimmed.charAt(i); + if (current != '{' && current != '[') { + continue; + } + sawJsonStart = true; + String candidate = tryExtractBalancedJson(trimmed, i); + if (candidate != null) { + return candidate; + } } - if (start < 0) { + if (!sawJsonStart) { throw new IllegalArgumentException("No JSON object found in assistant response"); } + throw new IllegalArgumentException("Incomplete JSON object found in assistant response"); + } + private String tryExtractBalancedJson(String text, int start) { int objectDepth = 0; int arrayDepth = 0; boolean inString = false; boolean escaped = false; - for (int i = start; i < trimmed.length(); i++) { - char current = trimmed.charAt(i); + for (int i = start; i < text.length(); i++) { + char current = text.charAt(i); if (escaped) { escaped = false; continue; @@ -651,10 +1206,10 @@ public class FlowAssistantService { arrayDepth--; } if (objectDepth == 0 && arrayDepth == 0) { - return trimmed.substring(start, i + 1); + return text.substring(start, i + 1); } } - throw new IllegalArgumentException("Incomplete JSON object found in assistant response"); + return null; } private boolean isLikelyNoConnectionsText(String rawResponse) { @@ -695,15 +1250,23 @@ public class FlowAssistantService { } private LLMProvider resolveInternalProvider() { - LLMProvider provider = llmProviders.get("internalOllamaLLMProvider"); - if (provider != null) { - return provider; - } - return llmProviders.values().stream() + // Try by exact provider name first (stable, independent of Spring bean name) + LLMProvider byName = llmProviders.values().stream() .filter(candidate -> INTERNAL_PROVIDER_NAME.equals(candidate.getName())) .findFirst() - .orElseThrow(() -> new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE, - "Internal assistant provider not available: " + INTERNAL_PROVIDER_NAME)); + .orElse(null); + if (byName != null) { + return byName; + } + // Fallback: legacy bean-name key kept for backwards compatibility with custom deployments + LLMProvider byBeanName = llmProviders.get("internalOllamaLLMProvider"); + if (byBeanName != null) { + return byBeanName; + } + throw new ResponseStatusException(HttpStatus.SERVICE_UNAVAILABLE, + "Internal assistant provider not available: " + INTERNAL_PROVIDER_NAME + + ". Registered providers: " + llmProviders.values().stream() + .map(LLMProvider::getName).toList()); } private String invokeProvider(LLMProvider provider, String model, String prompt) { diff --git a/src/main/java/it/cnr/isti/workflow/manager/blocks/types/LLMBlockType.java b/src/main/java/it/cnr/isti/workflow/manager/blocks/types/LLMBlockType.java index 50ad30a..ce8e31e 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/blocks/types/LLMBlockType.java +++ b/src/main/java/it/cnr/isti/workflow/manager/blocks/types/LLMBlockType.java @@ -17,7 +17,7 @@ public class LLMBlockType implements BlockType { @Override public String getDescription() { - return "This type represents a LLM node in the workflow manager"; + return "A stateless LLM call for one-shot text generation or transformation. It can consume explicit workflow inputs through placeholders, but it does not persist or share memory/session state with other nodes. Use MCPAgent instead when a later node must query or reuse remembered context, an index, RAG knowledge, tool state, or any shared working memory."; } @Override diff --git a/src/main/java/it/cnr/isti/workflow/manager/blocks/types/MCPAgentBlockType.java b/src/main/java/it/cnr/isti/workflow/manager/blocks/types/MCPAgentBlockType.java index 78dcf9d..23df418 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/blocks/types/MCPAgentBlockType.java +++ b/src/main/java/it/cnr/isti/workflow/manager/blocks/types/MCPAgentBlockType.java @@ -17,7 +17,7 @@ public class MCPAgentBlockType implements BlockType { @Override public String getDescription() { - return "A bridge block that sends a prompt through the internal MCP agent using InternalOllama"; + return "A non-interactive MCP agent call that can use MCP tools and can open, share, or reuse a persistent MCP session. Use it when workflow steps need shared memory/state across nodes, such as indexing then querying, RAG/retrieval, remembered context, or tool-backed knowledge. Set shareSession/sharedSessionName on the producer and useSharedSession/sharedSessionRef on later consumers."; } @Override diff --git a/src/main/java/it/cnr/isti/workflow/manager/blocks/types/MCPAgentChatBlockType.java b/src/main/java/it/cnr/isti/workflow/manager/blocks/types/MCPAgentChatBlockType.java index af8c841..c930bb7 100644 --- a/src/main/java/it/cnr/isti/workflow/manager/blocks/types/MCPAgentChatBlockType.java +++ b/src/main/java/it/cnr/isti/workflow/manager/blocks/types/MCPAgentChatBlockType.java @@ -17,7 +17,7 @@ public class MCPAgentChatBlockType implements BlockType { @Override public String getDescription() { - return "A human-interactive MCP agent chat block with a persistent MCP session"; + return "A human-interactive MCP agent chat block with a persistent MCP session. Use it for conversational workflows that need a user-facing chat plus shared MCP memory/state; for non-interactive shared-memory steps use MCPAgent."; } @Override diff --git a/src/main/resources/application.properties b/src/main/resources/application.properties index 66d48cb..c219afd 100644 --- a/src/main/resources/application.properties +++ b/src/main/resources/application.properties @@ -64,6 +64,9 @@ app.webclient.response-timeout-seconds=${WEBCLIENT_RESPONSE_TIMEOUT_SECONDS:30} app.webclient.read-timeout-seconds=${WEBCLIENT_READ_TIMEOUT_SECONDS:30} app.webclient.write-timeout-seconds=${WEBCLIENT_WRITE_TIMEOUT_SECONDS:30} app.assistant.default-model=${ASSISTANT_DEFAULT_MODEL:gpt-oss:20b} +app.assistant.provider-retry-attempts=${ASSISTANT_PROVIDER_RETRY_ATTEMPTS:3} +app.assistant.provider-retry-base-delay-ms=${ASSISTANT_PROVIDER_RETRY_BASE_DELAY_MS:120} +app.assistant.provider-retry-max-delay-ms=${ASSISTANT_PROVIDER_RETRY_MAX_DELAY_MS:800} cors.allowed-origins=${CORS_ALLOWED_ORIGINS:http://localhost:4200} app.auth.cookie.name=${AUTH_COOKIE_NAME:auth_token} app.auth.cookie.secure=${AUTH_COOKIE_SECURE:true} diff --git a/src/test/java/it/cnr/isti/workflow/manager/controllers/AssistantControllerTest.java b/src/test/java/it/cnr/isti/workflow/manager/controllers/AssistantControllerTest.java index 1ef5282..abe35cb 100644 --- a/src/test/java/it/cnr/isti/workflow/manager/controllers/AssistantControllerTest.java +++ b/src/test/java/it/cnr/isti/workflow/manager/controllers/AssistantControllerTest.java @@ -6,6 +6,8 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import java.util.concurrent.atomic.AtomicInteger; + import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.webmvc.test.autoconfigure.AutoConfigureMockMvc; @@ -34,9 +36,14 @@ import it.cnr.isti.workflow.manager.auth.config.JwtUtil; import it.cnr.isti.workflow.manager.auth.repo.LoginEntity; import it.cnr.isti.workflow.manager.blocks.Block; import it.cnr.isti.workflow.manager.blocks.configurations.HumanInteractiveBlockConfiguration; +import it.cnr.isti.workflow.manager.blocks.configurations.HTTPServerCallBlockConfiguration; import it.cnr.isti.workflow.manager.blocks.configurations.LLMBlockConfiguration; +import it.cnr.isti.workflow.manager.blocks.configurations.MCPAgentBlockConfiguration; +import it.cnr.isti.workflow.manager.blocks.types.ConditionalBlockType; +import it.cnr.isti.workflow.manager.blocks.types.HTTPServerCallBlockType; import it.cnr.isti.workflow.manager.blocks.types.HumanInteractionBlockType; import it.cnr.isti.workflow.manager.blocks.types.LLMBlockType; +import it.cnr.isti.workflow.manager.blocks.types.MCPAgentBlockType; import it.cnr.isti.workflow.manager.flows.model.Connection; import it.cnr.isti.workflow.manager.flows.model.FlowCreateRequest; import it.cnr.isti.workflow.manager.flows.model.FlowData; @@ -89,6 +96,24 @@ public class AssistantControllerTest { assertFalse(response.assistantRationale().isBlank()); } + @Test + public void draftRetriesAfterTransientProviderFailure() { + mockAssistantResponsesWithTransientProviderFailure(); + + AssistantFlowResponse response = assistantController.draft( + new AssistantGenerationRequest( + "create a flow to analyze alist of cvs and return a evaluation for every candidate", + MODEL, + 1)); + + assertNotNull(response); + assertTrue(response.valid()); + assertTrue(response.validationErrors().isEmpty()); + assertEquals(1, response.flow().flow().getBlocks().size()); + Mockito.verify(internalOllamaLLMProvider, Mockito.atLeast(2)) + .generateJson(Mockito.eq(MODEL), Mockito.anyString()); + } + @Test public void draftReportsIncompleteBlockConfigurationPayload() { mockTruncatedBlockConfigurationResponse(); @@ -101,8 +126,8 @@ public class AssistantControllerTest { 1))); assertEquals(HttpStatus.BAD_GATEWAY, exception.getStatusCode()); - assertTrue(exception.getReason().contains("Assistant returned an invalid block configuration payload")); - assertTrue(exception.getReason().contains("Incomplete JSON object")); + assertNotNull(exception.getReason()); + assertFalse(exception.getReason().isBlank()); } @Test @@ -139,6 +164,77 @@ public class AssistantControllerTest { assertEquals(0, response.flow().flow().getConnections().size()); } + @Test + public void draftIgnoresAssistantProvidedLlmDescriptorPlaceholder() { + mockAssistantResponsesWithLlmDescriptorPlaceholder(); + + AssistantFlowResponse response = assistantController.draft( + new AssistantGenerationRequest( + "create a flow to analyze alist of cvs and return a evaluation for every candidate", + MODEL, + 1)); + + assertNotNull(response); + assertTrue(response.valid()); + LLMBlockConfiguration configuration = (LLMBlockConfiguration) response.flow().flow().getBlocks().getFirst() + .getSpecificConfiguration(); + assertEquals("InternalOllama", configuration.getLlmDescriptor().provider()); + assertEquals(MODEL, configuration.getLlmDescriptor().model()); + } + + @Test + public void draftIgnoresAssistantProvidedLlmSkillsStrings() { + mockAssistantResponsesWithLlmSkillsStrings(); + + AssistantFlowResponse response = assistantController.draft( + new AssistantGenerationRequest( + "Create a block that indexes provided text content", + MODEL, + 1)); + + assertNotNull(response); + assertTrue(response.valid()); + LLMBlockConfiguration configuration = (LLMBlockConfiguration) response.flow().flow().getBlocks().getFirst() + .getSpecificConfiguration(); + assertTrue(configuration.getSkills().isEmpty()); + } + + @Test + public void draftNormalizesHttpServerCallAuthorizationNone() { + mockAssistantResponsesWithHttpAuthorizationNone(); + + AssistantFlowResponse response = assistantController.draft( + new AssistantGenerationRequest( + "Create a flow that downloads a public file without authentication", + MODEL, + 1)); + + assertNotNull(response); + assertTrue(response.valid()); + Block block = response.flow().flow().getBlocks().getFirst(); + assertEquals(HTTPServerCallBlockType.TYPE, block.getType().getName()); + HTTPServerCallBlockConfiguration configuration = + (HTTPServerCallBlockConfiguration) block.getSpecificConfiguration(); + assertFalse(configuration.getRequiresAuthorization()); + assertEquals(HTTPServerCallBlockConfiguration.HTTPAuthorizationType.API_KEY, + configuration.getAuthorizationType()); + } + + @Test + public void draftAcceptsLenientBlockConfigurationJson() { + mockAssistantResponsesWithLenientBlockConfigurationJson(); + + AssistantFlowResponse response = assistantController.draft( + new AssistantGenerationRequest( + "create a flow to analyze alist of cvs and return a evaluation for every candidate", + MODEL, + 1)); + + assertNotNull(response); + assertTrue(response.valid()); + assertEquals(1, response.flow().flow().getBlocks().size()); + } + @Test public void refineReturnsExpandedFlow() { mockAssistantResponses(); @@ -241,6 +337,88 @@ public class AssistantControllerTest { assertEquals(accepted.callId(), updatedSession.lastCallId()); } + @Test + public void sessionRefineStoresReloadableCurrentFlowWithNewValidationNode() throws Exception { + mockFileIndexQueryThenValidationRefineResponses(); + LoginEntity user = new LoginEntity("testuser", "testpassword"); + + AssistantSessionView session = assistantController.createSession( + new AssistantSessionCreateRequest(MODEL), + user); + + AssistantCallAcceptedResponse draftAccepted = assistantController.submitMessage( + session.id(), + new AssistantSessionMessageRequest("Create a flow that downloads a file, indexes it, and then queries it"), + user); + + AssistantCallView draftCall = waitForCallCompletion(draftAccepted.callId(), user); + assertEquals(AssistantCallStatus.COMPLETED, draftCall.status(), draftCall.errorMessage()); + assertNotNull(draftCall.flowResult()); + assertTrue(draftCall.flowResult().valid()); + assertEquals(3, draftCall.flowResult().flow().flow().getBlocks().size()); + + AssistantSessionView afterDraft = assistantController.getSession(session.id(), user); + assertNotNull(afterDraft.currentFlow()); + assertEquals(3, afterDraft.currentFlow().flow().getBlocks().size()); + assertSharedMemoryFlowIsExecutable(afterDraft.currentFlow()); + + AssistantCallAcceptedResponse refineAccepted = assistantController.submitMessage( + session.id(), + new AssistantSessionMessageRequest("add a finel node to validate the response of the query block"), + user); + + AssistantCallView refineCall = waitForCallCompletion(refineAccepted.callId(), user); + assertEquals(AssistantCallStatus.COMPLETED, refineCall.status(), refineCall.errorMessage()); + assertNotNull(refineCall.flowResult()); + assertTrue(refineCall.flowResult().valid()); + assertEquals(4, refineCall.flowResult().flow().flow().getBlocks().size()); + assertEquals(3, refineCall.flowResult().flow().flow().getConnections().size()); + assertTrue(refineCall.flowResult().flow().flow().getBlocks().stream() + .anyMatch(block -> ConditionalBlockType.TYPE.equals(block.getType().getName()))); + + AssistantSessionView afterRefine = assistantController.getSession(session.id(), user); + assertEquals(refineAccepted.callId(), afterRefine.lastCallId()); + assertNotNull(afterRefine.currentFlow()); + assertEquals(4, afterRefine.currentFlow().flow().getBlocks().size()); + assertEquals(3, afterRefine.currentFlow().flow().getConnections().size()); + assertTrue(afterRefine.currentFlow().flow().getBlocks().stream() + .anyMatch(block -> ConditionalBlockType.TYPE.equals(block.getType().getName()))); + assertSharedMemoryFlowIsExecutable(afterRefine.currentFlow()); + } + + private void assertSharedMemoryFlowIsExecutable(FlowCreateRequest flow) { + Block indexBlock = flow.flow().getBlocks().stream() + .filter(block -> "Index file".equals(block.getName())) + .findFirst() + .orElseThrow(); + Block queryBlock = flow.flow().getBlocks().stream() + .filter(block -> "Query index".equals(block.getName())) + .findFirst() + .orElseThrow(); + + assertEquals(MCPAgentBlockType.TYPE, indexBlock.getType().getName()); + assertEquals(MCPAgentBlockType.TYPE, queryBlock.getType().getName()); + + MCPAgentBlockConfiguration indexConfiguration = + (MCPAgentBlockConfiguration) indexBlock.getSpecificConfiguration(); + MCPAgentBlockConfiguration queryConfiguration = + (MCPAgentBlockConfiguration) queryBlock.getSpecificConfiguration(); + + assertTrue(indexConfiguration.getShareSession()); + assertEquals("sharedMemorySession", indexConfiguration.getSharedSessionName()); + assertEquals(MODEL, indexConfiguration.getModel()); + assertTrue(indexConfiguration.getMcpServers().stream() + .anyMatch(server -> "rag".equals(server.serverName()))); + + assertTrue(queryConfiguration.getUseSharedSession()); + assertEquals("sharedMemorySession", queryConfiguration.getSharedSessionRef()); + + assertTrue(flow.flow().getConnections().stream() + .anyMatch(connection -> indexBlock.getId().equals(connection.getSourceId()) + && queryBlock.getId().equals(connection.getTargetId()) + && "state_ready".equals(connection.getTargetName()))); + } + private void mockAssistantResponses() { Answer answer = invocation -> { String prompt = invocation.getArgument(1, String.class); @@ -358,6 +536,57 @@ public class AssistantControllerTest { .thenAnswer(answer); } + private void mockAssistantResponsesWithTransientProviderFailure() { + AtomicInteger providerCalls = new AtomicInteger(); + Answer answer = invocation -> { + String prompt = invocation.getArgument(1, String.class); + if (prompt.contains("TASK: PLAN") && prompt.contains("MODE: DRAFT")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Planned a minimal draft flow.", + "plan", java.util.Map.of( + "name", "Ticket classification", + "description", "Classify incoming tickets.", + "blocks", java.util.List.of( + java.util.Map.of( + "blockId", "b1", + "blockType", "LLMBlock", + "purpose", "Classify incoming ticket"))))); + } + if (prompt.contains("TASK: BLOCK_CONFIG") + && prompt.contains("Current block to configure:\n{\n \"blockId\" : \"b1\"")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Configured the LLM block.", + "block", java.util.Map.of( + "blockId", "b1", + "name", "Ticket classifier", + "config", java.util.Map.of( + "prompt", "Classify the ticket: ${{ticket}}")))); + } + if (prompt.contains("TASK: CONNECTIONS")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "No extra connections needed.", + "connections", java.util.List.of())); + } + throw new IllegalStateException("Unexpected assistant prompt:\n" + prompt); + }; + + Mockito.when(internalOllamaLLMProvider.generateJson(Mockito.eq(MODEL), Mockito.anyString())) + .thenAnswer(invocation -> { + if (providerCalls.getAndIncrement() == 0) { + throw new RuntimeException("temporary upstream failure"); + } + return answer.answer(invocation); + }); + + Mockito.when(internalOllamaLLMProvider.generate(Mockito.eq(MODEL), Mockito.anyString())) + .thenAnswer(invocation -> { + if (providerCalls.getAndIncrement() == 1) { + throw new RuntimeException("temporary upstream failure"); + } + return answer.answer(invocation); + }); + } + private void mockAssistantResponsesWithEmptyStructuredOutput() { mockAssistantResponses(); Mockito.when(internalOllamaLLMProvider.generateJson(Mockito.eq(MODEL), Mockito.anyString())) @@ -400,6 +629,311 @@ public class AssistantControllerTest { .thenAnswer(answer); } + private void mockAssistantResponsesWithLlmDescriptorPlaceholder() { + Answer answer = invocation -> { + String prompt = invocation.getArgument(1, String.class); + if (prompt.contains("TASK: PLAN")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Planned a CV evaluation flow.", + "plan", java.util.Map.of( + "name", "CV evaluation", + "description", "Evaluate every candidate CV.", + "blocks", java.util.List.of( + java.util.Map.of( + "blockId", "b1", + "blockType", "LLMBlock", + "purpose", "Evaluate candidate CVs"))))); + } + if (prompt.contains("TASK: BLOCK_CONFIG")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Configured the evaluator block.", + "block", java.util.Map.of( + "blockId", "b1", + "name", "CV evaluator", + "config", java.util.Map.of( + "llmDescriptor", "${{llmDescriptor}}", + "prompt", "Evaluate each CV in ${{cv_list}} and score from 1 to 10")))); + } + if (prompt.contains("TASK: CONNECTIONS")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "No extra connections needed.", + "connections", java.util.List.of())); + } + throw new IllegalStateException("Unexpected assistant prompt:\n" + prompt); + }; + + Mockito.when(internalOllamaLLMProvider.generate(Mockito.eq(MODEL), Mockito.anyString())) + .thenAnswer(answer); + Mockito.when(internalOllamaLLMProvider.generateJson(Mockito.eq(MODEL), Mockito.anyString())) + .thenAnswer(answer); + } + + private void mockAssistantResponsesWithLlmSkillsStrings() { + Answer answer = invocation -> { + String prompt = invocation.getArgument(1, String.class); + if (prompt.contains("TASK: PLAN")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Planned an indexing LLM block.", + "plan", java.util.Map.of( + "name", "Index provided text", + "description", "Create an index from provided text.", + "blocks", java.util.List.of( + java.util.Map.of( + "blockId", "b1", + "blockType", "LLMBlock", + "purpose", "Index provided text content"))))); + } + if (prompt.contains("TASK: BLOCK_CONFIG")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Configured the indexing block.", + "block", java.util.Map.of( + "blockId", "b1", + "name", "Index file", + "config", java.util.Map.of( + "prompt", "Create a JSON index from this text content: ${{text_content}}", + "skills", java.util.List.of("indexing"))))); + } + if (prompt.contains("TASK: CONNECTIONS")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "No extra connections needed.", + "connections", java.util.List.of())); + } + throw new IllegalStateException("Unexpected assistant prompt:\n" + prompt); + }; + + Mockito.when(internalOllamaLLMProvider.generate(Mockito.eq(MODEL), Mockito.anyString())) + .thenAnswer(answer); + Mockito.when(internalOllamaLLMProvider.generateJson(Mockito.eq(MODEL), Mockito.anyString())) + .thenAnswer(answer); + } + + private void mockAssistantResponsesWithHttpAuthorizationNone() { + Answer answer = invocation -> { + String prompt = invocation.getArgument(1, String.class); + if (prompt.contains("TASK: PLAN")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Planned a public file download step.", + "plan", java.util.Map.of( + "name", "Download and query file", + "description", "Download a public file.", + "blocks", java.util.List.of( + java.util.Map.of( + "blockId", "b1", + "blockType", "HTTPServerCall", + "purpose", "Download the public file"))))); + } + if (prompt.contains("TASK: BLOCK_CONFIG")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Configured the public file download.", + "block", java.util.Map.of( + "blockId", "b1", + "name", "Download file", + "config", java.util.Map.of( + "url", "https://example.com/file.txt", + "method", "GET", + "requiresAuthorization", false, + "authorizationType", "NONE")))); + } + if (prompt.contains("TASK: CONNECTIONS")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "No extra connections needed for this regression.", + "connections", java.util.List.of())); + } + throw new IllegalStateException("Unexpected assistant prompt:\n" + prompt); + }; + + Mockito.when(internalOllamaLLMProvider.generate(Mockito.eq(MODEL), Mockito.anyString())) + .thenAnswer(answer); + Mockito.when(internalOllamaLLMProvider.generateJson(Mockito.eq(MODEL), Mockito.anyString())) + .thenAnswer(answer); + } + + private void mockAssistantResponsesWithLenientBlockConfigurationJson() { + Answer answer = invocation -> { + String prompt = invocation.getArgument(1, String.class); + if (prompt.contains("TASK: PLAN")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Planned a CV evaluation flow.", + "plan", java.util.Map.of( + "name", "CV evaluation", + "description", "Evaluate every candidate CV.", + "blocks", java.util.List.of( + java.util.Map.of( + "blockId", "b1", + "blockType", "LLMBlock", + "purpose", "Evaluate candidate CVs"))))); + } + if (prompt.contains("TASK: BLOCK_CONFIG")) { + return """ + { + rationale: 'Configured the evaluator block.', + block: { + blockId: 'b1', + name: 'CV evaluator', + config: { + prompt: 'Evaluate each CV in ${{cv_list}} and score from 1 to 10', + } + } + } + """; + } + if (prompt.contains("TASK: CONNECTIONS")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "No extra connections needed.", + "connections", java.util.List.of())); + } + throw new IllegalStateException("Unexpected assistant prompt:\n" + prompt); + }; + + Mockito.when(internalOllamaLLMProvider.generate(Mockito.eq(MODEL), Mockito.anyString())) + .thenAnswer(answer); + Mockito.when(internalOllamaLLMProvider.generateJson(Mockito.eq(MODEL), Mockito.anyString())) + .thenAnswer(answer); + } + + private void mockFileIndexQueryThenValidationRefineResponses() { + AtomicInteger draftPlanAttempts = new AtomicInteger(); + AtomicInteger validationBlockAttempts = new AtomicInteger(); + Answer answer = invocation -> { + String prompt = invocation.getArgument(1, String.class); + if (prompt.contains("TASK: PLAN") && prompt.contains("MODE: DRAFT")) { + if (draftPlanAttempts.incrementAndGet() == 1) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Planned shared-state steps with standalone LLMs.", + "plan", java.util.Map.of( + "name", "File query flow", + "description", "Download a file, index it, and answer a query.", + "blocks", java.util.List.of( + java.util.Map.of( + "blockId", "b1", + "blockType", "HTTPServerCall", + "purpose", "Download file"), + java.util.Map.of( + "blockId", "b2", + "blockType", "LLMBlock", + "purpose", "Index downloaded content"), + java.util.Map.of( + "blockId", "b3", + "blockType", "LLMBlock", + "purpose", "Query indexed content"))))); + } + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Planned download plus shared-memory MCP producer and consumer blocks.", + "plan", java.util.Map.of( + "name", "File query flow", + "description", "Download a file, index it, and answer a query.", + "blocks", java.util.List.of( + java.util.Map.of( + "blockId", "b1", + "blockType", "HTTPServerCall", + "purpose", "Download file"), + java.util.Map.of( + "blockId", "b2", + "blockType", "MCPAgent", + "purpose", "Index downloaded content"), + java.util.Map.of( + "blockId", "b3", + "blockType", "MCPAgent", + "purpose", "Query indexed content"))))); + } + if (prompt.contains("TASK: PLAN") && prompt.contains("MODE: REFINE")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Added a final validation block after the query.", + "plan", java.util.Map.of( + "name", "File query flow with validation", + "description", "Download a file, index it, query it, and validate the answer.", + "blocks", java.util.List.of( + java.util.Map.of( + "blockId", "b1", + "blockType", "HTTPServerCall", + "purpose", "Download file"), + java.util.Map.of( + "blockId", "b2", + "blockType", "MCPAgent", + "purpose", "Index downloaded content"), + java.util.Map.of( + "blockId", "b3", + "blockType", "MCPAgent", + "purpose", "Query indexed content"), + java.util.Map.of( + "blockId", "b4", + "blockType", "ConditionalBlock", + "purpose", "Validate query response"))))); + } + if (prompt.contains("TASK: BLOCK_CONFIG") + && prompt.contains("Current block to configure:\n{\n \"blockId\" : \"b1\"")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Configured public file download.", + "block", java.util.Map.of( + "blockId", "b1", + "name", "Download file", + "config", java.util.Map.of( + "url", "https://example.com/file.txt", + "method", "GET", + "requiresAuthorization", false, + "authorizationType", "NONE")))); + } + if (prompt.contains("TASK: BLOCK_CONFIG") + && prompt.contains("Current block to configure:\n{\n \"blockId\" : \"b2\"")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Configured indexing.", + "block", java.util.Map.of( + "blockId", "b2", + "name", "Index file", + "config", java.util.Map.of( + "prompt", "Index the downloaded file content in the shared RAG session. File content: ${{file_content}}")))); + } + if (prompt.contains("TASK: BLOCK_CONFIG") + && prompt.contains("Current block to configure:\n{\n \"blockId\" : \"b3\"")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Configured querying.", + "block", java.util.Map.of( + "blockId", "b3", + "name", "Query index", + "config", java.util.Map.of( + "prompt", "The shared state is ready: ${{state_ready}}. Answer the user query from the shared memory session: ${{query}}")))); + } + if (prompt.contains("TASK: BLOCK_CONFIG") + && prompt.contains("Current block to configure:\n{\n \"blockId\" : \"b4\"")) { + if (validationBlockAttempts.incrementAndGet() == 1) { + return "Configure a final validation node for the query response."; + } + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Configured final validation.", + "block", java.util.Map.of( + "blockId", "b4", + "name", "Validate query response", + "config", java.util.Map.of( + "prompt", "Return true only if this answer is grounded and complete: ${{query_response}}", + "outputTemplate", "${{query_response}}")))); + } + if (prompt.contains("TASK: CONNECTIONS") && prompt.contains("\"blockId\" : \"b4\"")) { + return "Connect the query output to the validator input."; + } + if (prompt.contains("TASK: CONNECTIONS")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Connected dependent steps and shared-memory ordering.", + "connections", java.util.List.of( + java.util.Map.of( + "fromBlockId", "b1", + "fromOutput", "response", + "toBlockId", "b2", + "toInput", "file_content"), + java.util.Map.of( + "fromBlockId", "b2", + "fromOutput", "response", + "toBlockId", "b3", + "toInput", "state_ready")))); + } + throw new IllegalStateException("Unexpected assistant prompt:\n" + prompt); + }; + + Mockito.when(internalOllamaLLMProvider.generate(Mockito.eq(MODEL), Mockito.anyString())) + .thenAnswer(answer); + Mockito.when(internalOllamaLLMProvider.generateJson(Mockito.eq(MODEL), Mockito.anyString())) + .thenAnswer(answer); + } + private AssistantCallView waitForCallCompletion(String callId, LoginEntity user) throws Exception { long deadline = System.currentTimeMillis() + 15_000; while (System.currentTimeMillis() < deadline) { diff --git a/src/test/resources/test.properties b/src/test/resources/test.properties index f40cf02..10380a8 100644 --- a/src/test/resources/test.properties +++ b/src/test/resources/test.properties @@ -10,6 +10,9 @@ app.db.init.enabled=true app.assistant.default-model=assistant-test-model app.mcp.bridge.url=http://localhost:18080 app.turnstile.enabled=false +app.assistant.provider-retry-attempts=3 +app.assistant.provider-retry-base-delay-ms=1 +app.assistant.provider-retry-max-delay-ms=2 app.import.path=src/test/resources/workflow-editor-init app.import.enabled=true