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 87c7bfc..74616cc 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 @@ -91,6 +91,17 @@ public class FlowAssistantPromptService { "purpose": "...", "operation": "ADD" } + ], + "containers": [ + { + "containerId": "c1", + "containerType": "GenericContainer", + "purpose": "...", + "operation": "ADD", + "blocks": [ + { "blockId": "c1-b1", "blockType": "LLMBlock", "purpose": "..." } + ] + } ] } } @@ -104,14 +115,17 @@ public class FlowAssistantPromptService { - Prefer HTTPServerCall for fetching files, APIs, or URLs before processing them. - If a task can be completed with a single stateless step, do not introduce MCPAgent. - If a later step must reuse data from an earlier step, make the earlier step produce it and the later step consume it explicitly. - - For DRAFT with a non-empty user request, "blocks" must contain at least one block. - - blockId must be stable symbolic ids like b1, b2, b3. + - For DRAFT with a non-empty user request, "blocks" and "containers" combined must contain at least one entry. + - blockId must be stable symbolic ids like b1, b2, b3. containerId must be stable symbolic ids like c1, c2, and must not collide with any blockId. - operation must be one of KEEP, ADD, UPDATE, REMOVE. - - For DRAFT, use ADD for every block. - - For REFINE/FIX with a current flow, mark unchanged existing blocks as KEEP. - - Use ADD only for new blocks, UPDATE only for existing blocks whose configuration must change, and REMOVE only for existing blocks to delete. - - For KEEP blocks, reuse the current block id/name when available and do not imply configuration changes. - - Do not return block configuration yet. + - For DRAFT, use ADD for every block and container. + - For REFINE/FIX with a current flow, mark unchanged existing blocks and containers as KEEP. + - Use ADD only for new blocks/containers, UPDATE only for existing ones whose configuration must change, and REMOVE only for existing ones to delete. + - For KEEP blocks/containers, reuse the current id/name when available and do not imply configuration changes. + - "containers" is optional; omit it (or leave it empty) unless grouping is actually useful. + - GROUPING RULE: put a cohesive group of blocks (e.g. 3+ steps that form one logical phase) inside a "containers" entry instead of leaving them all as flat top-level blocks, when doing so makes the flow's overall structure clearer - especially for flows with many steps across several logical phases. A container's own "blocks" list uses the exact same block shape as top-level blocks (blockId/blockType/purpose), but do not set "operation" on inner blocks and never put another container inside a container (nesting is not supported) - a container always needs at least one inner block. + - containerType must currently be exactly "GenericContainer" (a plain grouping container with no looping/iteration semantics); do not invent other container types. + - Do not return block or container configuration yet. - Do not return connections yet. - Return valid JSON with no markdown fences. @@ -284,6 +298,7 @@ public class FlowAssistantPromptService { - If no connections are needed, still return JSON with "connections": []. - Every input declared on a BranchRejoinBlock must receive its own connection from a different upstream branch output; never leave one unconnected. - A block whose text only references ${{global.name}} has no matching input to connect to for that value - it is resolved directly from the flow-level global input, not from a connection. Do not invent a connection for it and do not treat it as an unconnected input needing a producer. + - Containers (containerType GenericContainer) are valid connection endpoints exactly like blocks: use their containerId as fromBlockId/toBlockId, and their listed inputs/outputs (auto-derived from whatever their inner blocks leave unconnected) as fromOutput/toInput. Never reference an inner block of a container directly from outside that container - only the container's own exposed inputs/outputs are reachable from the rest of the flow. - Return valid JSON with no markdown fences. EXAMPLE (illustrative only; block ids are placeholders for whatever ids the flow plan actually uses): 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 b68b43d..fd61179 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 @@ -44,6 +44,11 @@ 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.containers.Container; +import it.cnr.isti.workflow.manager.containers.configurations.GenericContainerConfiguration; +import it.cnr.isti.workflow.manager.containers.factories.GenericContainerFactory; +import it.cnr.isti.workflow.manager.containers.types.GenericContainerType; +import it.cnr.isti.workflow.manager.flows.model.FlowNode; 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; @@ -150,12 +155,26 @@ public class FlowAssistantService { private static final ProgressListener NOOP_PROGRESS = (phase, message) -> { }; - private record AssistantFlowPlan(String name, String description, List blocks) { + private record AssistantFlowPlan(String name, String description, List blocks, + List containers) { + private AssistantFlowPlan(String name, String description, List blocks) { + this(name, description, blocks, List.of()); + } } private record AssistantBlockPlan(String blockId, String blockType, String purpose, String operation) { } + /** + * A GenericContainer grouping in the plan: an inner subflow of its own blocks (never + * containers - nesting isn't supported). Only "GenericContainer" is a valid containerType + * for now; IteratorContainer/LoopContainer need iteration/guard semantics the assistant + * doesn't author yet. + */ + private record AssistantContainerPlan(String containerId, String containerType, String purpose, String operation, + List blocks) { + } + private enum PlanOperation { KEEP, ADD, @@ -209,6 +228,9 @@ public class FlowAssistantService { @Autowired private FlowExecutionValidator flowExecutionValidator; + @Autowired + private GenericContainerFactory genericContainerFactory; + @Value("${app.assistant.provider-retry-base-delay-ms:" + DEFAULT_RETRY_BASE_DELAY_MILLIS + "}") private long retryBaseDelayMillis; @@ -369,12 +391,14 @@ public class FlowAssistantService { boolean requireSharedMemorySemantics = isSharedMemoryContext(userPrompt, currentFlow, parsedPlan.plan()); List> assembledBlocks = new ArrayList<>(); - Map> blocksByPlanId = new LinkedHashMap<>(); - Map> blocksByAlias = new LinkedHashMap<>(); - Map> oldBlockIdToAssembledBlock = new LinkedHashMap<>(); - Set removedExistingBlockIds = new LinkedHashSet<>(); + List> assembledContainers = new ArrayList<>(); + Map nodesByPlanId = new LinkedHashMap<>(); + Map nodesByAlias = new LinkedHashMap<>(); + Map oldNodeIdToAssembledNode = new LinkedHashMap<>(); + Set removedExistingNodeIds = new LinkedHashSet<>(); Set usedExistingBlockIds = new LinkedHashSet<>(); - List configuredBlocks = new ArrayList<>(); + Set usedExistingContainerIds = new LinkedHashSet<>(); + List configuredNodes = new ArrayList<>(); List rationaleParts = new ArrayList<>(); appendRationale(rationaleParts, parsedPlan.rationale()); @@ -389,7 +413,7 @@ public class FlowAssistantService { if (operation == PlanOperation.REMOVE) { if (existingBlock != null) { - removedExistingBlockIds.add(existingBlock.getId()); + removedExistingNodeIds.add(existingBlock.getId()); } continue; } @@ -425,14 +449,14 @@ public class FlowAssistantService { } assembledBlocks.add(assembledBlock); - blocksByPlanId.put(blockPlan.blockId(), assembledBlock); + nodesByPlanId.put(blockPlan.blockId(), assembledBlock); if (existingBlock != null) { - oldBlockIdToAssembledBlock.put(existingBlock.getId(), assembledBlock); + oldNodeIdToAssembledNode.put(existingBlock.getId(), assembledBlock); } - registerBlockAlias(blocksByAlias, blockPlan.blockId(), assembledBlock); - registerBlockAlias(blocksByAlias, assembledBlock.getName(), assembledBlock); - registerBlockAlias(blocksByAlias, blockPlan.purpose(), assembledBlock); - configuredBlocks.add(new ConfiguredBlockSummary( + registerNodeAlias(nodesByAlias, blockPlan.blockId(), assembledBlock); + registerNodeAlias(nodesByAlias, assembledBlock.getName(), assembledBlock); + registerNodeAlias(nodesByAlias, blockPlan.purpose(), assembledBlock); + configuredNodes.add(new ConfiguredBlockSummary( blockPlan.blockId(), blockPlan.blockType(), assembledBlock.getName(), @@ -443,34 +467,81 @@ public class FlowAssistantService { : assembledBlock.getOutputs().stream().map(io -> io.getName()).toList())); } + for (AssistantContainerPlan containerPlan : parsedPlan.plan().containers() == null + ? List.of() + : parsedPlan.plan().containers()) { + PlanOperation containerOperation = parsePlanOperation(containerPlan.operation()); + Container existingContainer = resolveExistingContainerForPlan(containerPlan, currentFlow, -1, + usedExistingContainerIds); + if (existingContainer != null) { + usedExistingContainerIds.add(existingContainer.getId()); + } + + if (containerOperation == PlanOperation.REMOVE) { + if (existingContainer != null) { + removedExistingNodeIds.add(existingContainer.getId()); + } + continue; + } + + Container assembledContainer; + if (containerOperation == PlanOperation.KEEP && existingContainer != null) { + progressListener.onProgress("configuring_blocks", "Keeping container " + containerPlan.containerId()); + assembledContainer = existingContainer; + } else { + progressListener.onProgress("configuring_blocks", "Assembling container " + containerPlan.containerId()); + assembledContainer = assembleGenericContainer(provider, workflowModel, phaseModels, mode, userPrompt, + containerPlan, catalogByType, rationaleParts, progressListener); + } + + assembledContainers.add(assembledContainer); + nodesByPlanId.put(containerPlan.containerId(), assembledContainer); + if (existingContainer != null) { + oldNodeIdToAssembledNode.put(existingContainer.getId(), assembledContainer); + } + registerNodeAlias(nodesByAlias, containerPlan.containerId(), assembledContainer); + registerNodeAlias(nodesByAlias, assembledContainer.getName(), assembledContainer); + registerNodeAlias(nodesByAlias, containerPlan.purpose(), assembledContainer); + configuredNodes.add(new ConfiguredBlockSummary( + containerPlan.containerId(), + containerPlan.containerType(), + assembledContainer.getName(), + containerPlan.purpose(), + assembledContainer.getInputs() == null ? List.of() + : assembledContainer.getInputs().stream().map(io -> io.getName()).toList(), + assembledContainer.getOutputs() == null ? List.of() + : assembledContainer.getOutputs().stream().map(io -> io.getName()).toList())); + } + progressListener.onProgress("connecting_blocks", "Connecting configured blocks"); + int assembledNodeCount = assembledBlocks.size() + assembledContainers.size(); ParsedConnections parsedConnections; if (targetedRepair) { parsedConnections = new ParsedConnections(List.of(), "Targeted repair: only the flagged blocks were reconfigured, connections are unchanged."); - } else if (assembledBlocks.size() < 2) { - parsedConnections = new ParsedConnections(List.of(), "No connections needed for a single-block flow."); + } else if (assembledNodeCount < 2) { + parsedConnections = new ParsedConnections(List.of(), "No connections needed for a single-node flow."); } else { String connectionsPrompt = promptService.buildConnectionsPrompt(mode, userPrompt, parsedPlan.plan(), - configuredBlocks, currentFlow, errors); + configuredNodes, currentFlow, errors); parsedConnections = invokeStructuredAndValidate(provider, jsonModelFor(mode, phaseModels), phaseModels.repairModel(), connectionsPrompt, "connections", rawResponse -> { ParsedConnections parsed = parseConnectionsOrInferSequential(rawResponse, assembledBlocks); parsed = completeRequiredSequentialConnections(requireSharedMemorySemantics, parsed, - assembledBlocks, blocksByPlanId, blocksByAlias); - List candidateConnections = toValidConnections(parsed.connections(), blocksByPlanId, - blocksByAlias); - validateSharedMemorySemantics(requireSharedMemorySemantics, blocksByPlanId.values(), + assembledBlocks, nodesByPlanId, nodesByAlias); + List candidateConnections = toValidConnections(parsed.connections(), nodesByPlanId, + nodesByAlias); + validateSharedMemorySemantics(requireSharedMemorySemantics, assembledBlocks, candidateConnections); return parsed; }); } appendRationale(rationaleParts, parsedConnections.rationale()); - List generatedConnections = toValidConnections(parsedConnections.connections(), blocksByPlanId, - blocksByAlias); + List generatedConnections = toValidConnections(parsedConnections.connections(), nodesByPlanId, + nodesByAlias); List connections = mergeConnections( - preserveCurrentConnections(currentFlow, oldBlockIdToAssembledBlock, removedExistingBlockIds), + preserveCurrentConnections(currentFlow, oldNodeIdToAssembledNode, removedExistingNodeIds), generatedConnections); FlowCreateRequest flow = new FlowCreateRequest( @@ -478,6 +549,7 @@ public class FlowAssistantService { defaultIfBlank(parsedPlan.plan().description(), currentFlow == null ? null : currentFlow.description()), FlowData.builder() .blocks(assembledBlocks) + .containers(assembledContainers) .connections(connections) .globalInputs(collectGlobalInputs(assembledBlocks, currentFlow)) .build()); @@ -485,6 +557,101 @@ public class FlowAssistantService { return new AssembledFlow(flow, String.join(" ", rationaleParts).trim()); } + /** + * Assembles a GenericContainer from a container plan entry: configures each inner block + * (always freshly - a container's own subflow isn't incrementally diffed across FIX/REFINE + * rounds in this version), wires them together, then lets GenericContainerFactory derive the + * container's exposed inputs/outputs from whichever inner handles are left unconnected. + */ + private Container assembleGenericContainer(LLMProvider provider, String workflowModel, + ResolvedAssistantModels phaseModels, OperationMode mode, String userPrompt, + AssistantContainerPlan containerPlan, Map catalogByType, + List rationaleParts, ProgressListener progressListener) { + List> innerBlocks = new ArrayList<>(); + Map innerNodesByPlanId = new LinkedHashMap<>(); + Map innerNodesByAlias = new LinkedHashMap<>(); + List innerConfiguredNodes = new ArrayList<>(); + + List innerBlockPlans = containerPlan.blocks(); + for (int blockIndex = 0; blockIndex < innerBlockPlans.size(); blockIndex++) { + AssistantBlockPlan blockPlan = innerBlockPlans.get(blockIndex); + BlockCatalogService.AssistantPromptBlockDescriptor descriptor = catalogByType.get(blockPlan.blockType()); + if (descriptor == null) { + throw new ResponseStatusException(HttpStatus.BAD_GATEWAY, + "Assistant selected an unknown block type inside container " + containerPlan.containerId() + + ": " + blockPlan.blockType()); + } + progressListener.onProgress("configuring_blocks", + "Configuring block " + blockPlan.blockId() + " in container " + containerPlan.containerId()); + String blockPrompt = promptService.buildBlockConfigurationPrompt(mode, userPrompt, descriptor, + containerInnerPlan(containerPlan), blockPlan, null, List.of(), workflowModel); + int currentBlockIndex = blockIndex; + int blockPlanCount = innerBlockPlans.size(); + ConfiguredBlockResult configuredBlock = invokeStructuredAndValidate(provider, + jsonModelFor(mode, phaseModels), phaseModels.repairModel(), blockPrompt, + "block configuration for " + blockPlan.blockId() + " in container " + containerPlan.containerId(), + rawResponse -> { + ParsedBlockDraft parsedBlock = parseBlockDraft(rawResponse); + AssistantConfiguredBlockDraft normalizedDraft = normalizeBlockDraft(blockPlan, parsedBlock.block()); + Block newBlock = buildBlock(descriptor, blockPlan, normalizedDraft, workflowModel, false, + currentBlockIndex, blockPlanCount); + return new ConfiguredBlockResult(parsedBlock, newBlock); + }); + appendRationale(rationaleParts, configuredBlock.parsedBlock().rationale()); + Block assembledBlock = configuredBlock.block(); + + innerBlocks.add(assembledBlock); + innerNodesByPlanId.put(blockPlan.blockId(), assembledBlock); + registerNodeAlias(innerNodesByAlias, blockPlan.blockId(), assembledBlock); + registerNodeAlias(innerNodesByAlias, assembledBlock.getName(), assembledBlock); + registerNodeAlias(innerNodesByAlias, blockPlan.purpose(), assembledBlock); + innerConfiguredNodes.add(new ConfiguredBlockSummary( + blockPlan.blockId(), + blockPlan.blockType(), + assembledBlock.getName(), + blockPlan.purpose(), + assembledBlock.getInputs() == null ? List.of() + : assembledBlock.getInputs().stream().map(io -> io.getName()).toList(), + assembledBlock.getOutputs() == null ? List.of() + : assembledBlock.getOutputs().stream().map(io -> io.getName()).toList())); + } + + ParsedConnections innerConnections; + if (innerBlocks.size() < 2) { + innerConnections = new ParsedConnections(List.of(), "No connections needed inside this container."); + } else { + String connectionsPrompt = promptService.buildConnectionsPrompt(mode, userPrompt, + containerInnerPlan(containerPlan), innerConfiguredNodes, null, List.of()); + innerConnections = invokeStructuredAndValidate(provider, jsonModelFor(mode, phaseModels), + phaseModels.repairModel(), connectionsPrompt, + "connections for container " + containerPlan.containerId(), rawResponse -> { + ParsedConnections parsed = parseConnectionsOrInferSequential(rawResponse, innerBlocks); + toValidConnections(parsed.connections(), innerNodesByPlanId, innerNodesByAlias); + return parsed; + }); + } + appendRationale(rationaleParts, innerConnections.rationale()); + List connections = toValidConnections(innerConnections.connections(), innerNodesByPlanId, + innerNodesByAlias); + + GenericContainerConfiguration configuration = GenericContainerConfiguration.builder() + .name(defaultIfBlank(containerPlan.purpose(), containerPlan.containerId())) + .subFlow(FlowData.builder() + .blocks(innerBlocks) + .connections(connections) + .build()) + .build(); + return genericContainerFactory.create(configuration); + } + + private AssistantFlowPlan containerInnerPlan(AssistantContainerPlan containerPlan) { + return new AssistantFlowPlan( + defaultIfBlank(containerPlan.purpose(), containerPlan.containerId()), + containerPlan.purpose(), + containerPlan.blocks(), + List.of()); + } + /** * Auto-declares a global input for every distinct ${{global.x}} reference found across the * assembled blocks' configuration, preserving any pre-existing declaration (type/multiplicity) @@ -967,7 +1134,11 @@ public class FlowAssistantService { * so the PLAN and CONNECTIONS phases can be skipped for this FIX round. */ private boolean isTargetedBlockRepairEligible(FlowCreateRequest currentFlow, List errors) { - if (errors == null || errors.isEmpty() || !hasCurrentFlowBlocks(currentFlow)) { + // Reusing the plan verbatim (buildReusedPlanForTargetedRepair) only reconstructs the + // block list - it would silently drop any existing containers. Bail out to the full + // repair path whenever containers are present rather than risk that. + if (errors == null || errors.isEmpty() || !hasCurrentFlowBlocks(currentFlow) + || hasCurrentFlowContainers(currentFlow)) { return false; } Set existingBlockIds = currentFlow.flow().getBlocks().stream() @@ -1064,7 +1235,7 @@ public class FlowAssistantService { } private List preserveCurrentConnections(FlowCreateRequest currentFlow, - Map> oldBlockIdToAssembledBlock, Set removedExistingBlockIds) { + Map oldNodeIdToAssembledNode, Set removedExistingNodeIds) { if (currentFlow == null || currentFlow.flow() == null || currentFlow.flow().getConnections() == null || currentFlow.flow().getConnections().isEmpty()) { return List.of(); @@ -1073,12 +1244,12 @@ public class FlowAssistantService { List preserved = new ArrayList<>(); for (Connection connection : currentFlow.flow().getConnections()) { if (connection == null - || removedExistingBlockIds.contains(connection.getSourceId()) - || removedExistingBlockIds.contains(connection.getTargetId())) { + || removedExistingNodeIds.contains(connection.getSourceId()) + || removedExistingNodeIds.contains(connection.getTargetId())) { continue; } - Block source = oldBlockIdToAssembledBlock.get(connection.getSourceId()); - Block target = oldBlockIdToAssembledBlock.get(connection.getTargetId()); + FlowNode source = oldNodeIdToAssembledNode.get(connection.getSourceId()); + FlowNode target = oldNodeIdToAssembledNode.get(connection.getTargetId()); if (source == null || target == null) { continue; } @@ -1112,7 +1283,7 @@ public class FlowAssistantService { } private List toValidConnections(List draftedConnections, - Map> blocksByPlanId, Map> blocksByAlias) { + Map nodesByPlanId, Map nodesByAlias) { if (draftedConnections == null || draftedConnections.isEmpty()) { return List.of(); } @@ -1120,7 +1291,7 @@ public class FlowAssistantService { List validConnections = new ArrayList<>(); for (AssistantConnectionDraft draftedConnection : draftedConnections) { try { - validConnections.add(toConnection(draftedConnection, blocksByPlanId, blocksByAlias)); + validConnections.add(toConnection(draftedConnection, nodesByPlanId, nodesByAlias)); } catch (ResponseStatusException e) { if (!HttpStatus.BAD_GATEWAY.equals(e.getStatusCode())) { throw e; @@ -1133,22 +1304,22 @@ public class FlowAssistantService { return List.copyOf(validConnections); } - private Connection toConnection(AssistantConnectionDraft connection, Map> blocksByPlanId, - Map> blocksByAlias) { - Block source = resolveConnectionBlock(connection.fromBlockId(), blocksByPlanId, blocksByAlias); - Block target = resolveConnectionBlock(connection.toBlockId(), blocksByPlanId, blocksByAlias); + private Connection toConnection(AssistantConnectionDraft connection, Map nodesByPlanId, + Map nodesByAlias) { + FlowNode source = resolveConnectionBlock(connection.fromBlockId(), nodesByPlanId, nodesByAlias); + FlowNode target = resolveConnectionBlock(connection.toBlockId(), nodesByPlanId, nodesByAlias); if (source == null) { - source = inferBlockByIo(connection.fromOutput(), blocksByPlanId.values(), true); + source = inferBlockByIo(connection.fromOutput(), nodesByPlanId.values(), true); } if (target == null) { - target = inferBlockByIo(connection.toInput(), blocksByPlanId.values(), false); + target = inferBlockByIo(connection.toInput(), nodesByPlanId.values(), false); } if (source == null || target == null) { throw new ResponseStatusException(HttpStatus.BAD_GATEWAY, "Assistant returned a connection with unknown block ids" + " (fromBlockId=" + connection.fromBlockId() + ", toBlockId=" + connection.toBlockId() - + ", allowedBlockIds=" + blocksByPlanId.keySet() + ")"); + + ", allowedBlockIds=" + nodesByPlanId.keySet() + ")"); } String sourceName = resolveConnectionOutputName(source, connection.fromOutput()); String targetName = resolveConnectionInputName(target, connection.toInput()); @@ -1208,7 +1379,7 @@ public class FlowAssistantService { } private ParsedConnections completeRequiredSequentialConnections(boolean required, ParsedConnections parsed, - List> assembledBlocks, Map> blocksByPlanId, Map> blocksByAlias) { + List> assembledBlocks, Map nodesByPlanId, Map nodesByAlias) { if (!required) { return parsed; } @@ -1216,15 +1387,15 @@ public class FlowAssistantService { 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); + FlowNode inferredSource = resolveConnectionBlock(inferred.fromBlockId(), nodesByPlanId, nodesByAlias); + FlowNode inferredTarget = resolveConnectionBlock(inferred.toBlockId(), nodesByPlanId, nodesByAlias); 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); + FlowNode existingSource = resolveConnectionBlock(existing.fromBlockId(), nodesByPlanId, nodesByAlias); + FlowNode existingTarget = resolveConnectionBlock(existing.toBlockId(), nodesByPlanId, nodesByAlias); if (existingSource != null && existingTarget != null && Objects.equals(existingSource.getId(), inferredSource.getId()) && Objects.equals(existingTarget.getId(), inferredTarget.getId())) { @@ -1264,26 +1435,26 @@ public class FlowAssistantService { .contains(normalized); } - private Block resolveConnectionBlock(String rawReference, Map> blocksByPlanId, - Map> blocksByAlias) { + private FlowNode resolveConnectionBlock(String rawReference, Map nodesByPlanId, + Map nodesByAlias) { if (rawReference == null || rawReference.isBlank()) { return null; } - Block direct = blocksByPlanId.get(rawReference); + FlowNode direct = nodesByPlanId.get(rawReference); if (direct != null) { return direct; } - return blocksByAlias.get(normalizeBlockReference(rawReference)); + return nodesByAlias.get(normalizeBlockReference(rawReference)); } - private Block inferBlockByIo(String ioName, Collection> blocks, boolean output) { + private FlowNode inferBlockByIo(String ioName, Collection nodes, boolean output) { String normalizedIo = normalizeBlockReference(ioName); if (normalizedIo == null) { return null; } - Block match = null; - for (Block block : blocks) { - List ioDescriptors = output ? block.getOutputs() : block.getInputs(); + FlowNode match = null; + for (FlowNode node : nodes) { + List ioDescriptors = output ? node.getOutputs() : node.getInputs(); if (ioDescriptors == null) { continue; } @@ -1295,17 +1466,17 @@ public class FlowAssistantService { if (match != null) { return null; } - match = block; + match = node; } return match; } - private String resolveConnectionOutputName(Block block, String requestedOutput) { - return resolveIoName(block.getOutputs(), requestedOutput, List.of("response", "output", "true", "false")); + private String resolveConnectionOutputName(FlowNode node, String requestedOutput) { + return resolveIoName(node.getOutputs(), requestedOutput, List.of("response", "output", "true", "false")); } - private String resolveConnectionInputName(Block block, String requestedInput) { - return resolveIoName(block.getInputs(), requestedInput, List.of("input", "prompt")); + private String resolveConnectionInputName(FlowNode node, String requestedInput) { + return resolveIoName(node.getInputs(), requestedInput, List.of("input", "prompt")); } private String resolveIoName(List descriptors, String requestedName, List preferredNames) { @@ -1341,10 +1512,10 @@ public class FlowAssistantService { .orElse(null); } - private void registerBlockAlias(Map> blocksByAlias, String reference, Block block) { + private void registerNodeAlias(Map nodesByAlias, String reference, FlowNode node) { String normalized = normalizeBlockReference(reference); if (normalized != null) { - blocksByAlias.putIfAbsent(normalized, block); + nodesByAlias.putIfAbsent(normalized, node); } } @@ -1505,7 +1676,78 @@ public class FlowAssistantService { block.purpose(), operation.name())); } - return new AssistantFlowPlan(plan.name(), plan.description(), normalizedBlocks); + + List normalizedContainers = new ArrayList<>(); + Set usedExistingContainerIds = new LinkedHashSet<>(); + List rawContainers = plan.containers() == null ? List.of() : plan.containers(); + for (int containerIndex = 0; containerIndex < rawContainers.size(); containerIndex++) { + AssistantContainerPlan containerPlan = rawContainers.get(containerIndex); + if (containerPlan.containerId() == null || containerPlan.containerId().isBlank() + || !ids.add(containerPlan.containerId())) { + throw new ResponseStatusException(HttpStatus.BAD_GATEWAY, + "Assistant returned invalid or duplicate container ids in the plan"); + } + if (!GenericContainerType.TYPE.equals(containerPlan.containerType())) { + throw new ResponseStatusException(HttpStatus.BAD_GATEWAY, + "Assistant selected an unsupported container type: " + containerPlan.containerType() + + " (only " + GenericContainerType.TYPE + " can currently be authored by the assistant)"); + } + PlanOperation containerOperation = normalizeContainerOperation(containerPlan, mode, currentFlow, errors, + containerIndex, usedExistingContainerIds); + List innerBlocks = containerOperation == PlanOperation.KEEP + || containerOperation == PlanOperation.REMOVE + ? List.of() + : normalizeContainerInnerBlocks(containerPlan); + normalizedContainers.add(new AssistantContainerPlan( + containerPlan.containerId(), + containerPlan.containerType(), + containerPlan.purpose(), + containerOperation.name(), + innerBlocks)); + } + + return new AssistantFlowPlan(plan.name(), plan.description(), normalizedBlocks, normalizedContainers); + } + + /** + * Container inner blocks are always fully (re)generated on ADD/UPDATE - the assistant does + * not incrementally diff a container's own subflow across FIX/REFINE rounds in this version, + * so any KEEP/UPDATE/REMOVE the model assigns to an inner block is ignored in favor of ADD. + */ + private List normalizeContainerInnerBlocks(AssistantContainerPlan containerPlan) { + List rawInnerBlocks = containerPlan.blocks() == null ? List.of() : containerPlan.blocks(); + if (rawInnerBlocks.isEmpty()) { + throw new ResponseStatusException(HttpStatus.BAD_GATEWAY, + "Assistant returned a container with no inner blocks: " + containerPlan.containerId()); + } + Set innerIds = new LinkedHashSet<>(); + List innerBlocks = new ArrayList<>(); + for (AssistantBlockPlan innerBlock : rawInnerBlocks) { + if (innerBlock.blockId() == null || innerBlock.blockId().isBlank() || !innerIds.add(innerBlock.blockId())) { + throw new ResponseStatusException(HttpStatus.BAD_GATEWAY, + "Assistant returned invalid or duplicate inner block ids in container: " + + containerPlan.containerId()); + } + if (innerBlock.blockType() == null || innerBlock.blockType().isBlank()) { + throw new ResponseStatusException(HttpStatus.BAD_GATEWAY, + "Assistant returned an inner block without blockType in container: " + + containerPlan.containerId()); + } + if (isContainerBlockType(innerBlock.blockType())) { + throw new ResponseStatusException(HttpStatus.BAD_GATEWAY, + "Nested containers are not supported: container " + containerPlan.containerId() + + " cannot contain another container"); + } + innerBlocks.add(new AssistantBlockPlan(innerBlock.blockId(), innerBlock.blockType(), innerBlock.purpose(), + PlanOperation.ADD.name())); + } + return innerBlocks; + } + + private boolean isContainerBlockType(String blockType) { + return GenericContainerType.TYPE.equals(blockType) + || "IteratorContainer".equals(blockType) + || "LoopContainer".equals(blockType); } private PlanOperation normalizePlanOperation(AssistantBlockPlan block, OperationMode mode, @@ -1549,6 +1791,109 @@ public class FlowAssistantService { return false; } + private PlanOperation normalizeContainerOperation(AssistantContainerPlan containerPlan, OperationMode mode, + FlowCreateRequest currentFlow, List errors, int containerIndex, + Set usedExistingContainerIds) { + if (containerPlan.operation() != null && !containerPlan.operation().isBlank()) { + PlanOperation explicitOperation = parsePlanOperation(containerPlan.operation()); + if (explicitOperation != PlanOperation.ADD) { + Container matched = resolveExistingContainerForPlan(containerPlan, currentFlow, containerIndex, + usedExistingContainerIds); + if (matched != null) { + usedExistingContainerIds.add(matched.getId()); + } + } + return explicitOperation; + } + if (mode == OperationMode.DRAFT || !hasCurrentFlowContainers(currentFlow)) { + return PlanOperation.ADD; + } + + Container matched = resolveExistingContainerForPlan(containerPlan, currentFlow, containerIndex, + usedExistingContainerIds); + if (matched == null) { + return PlanOperation.ADD; + } + usedExistingContainerIds.add(matched.getId()); + if (mode == OperationMode.FIX && hasValidationErrorForContainer(matched, errors)) { + return PlanOperation.UPDATE; + } + return PlanOperation.KEEP; + } + + private boolean hasValidationErrorForContainer(Container container, List errors) { + if (container == null || container.getId() == null || errors == null || errors.isEmpty()) { + return false; + } + for (ValidationError error : errors) { + if (Objects.equals(container.getId(), error.id()) + || error.relatedNodeIds() != null && error.relatedNodeIds().contains(container.getId())) { + return true; + } + } + return false; + } + + private boolean hasCurrentFlowContainers(FlowCreateRequest currentFlow) { + return currentFlow != null + && currentFlow.flow() != null + && currentFlow.flow().getContainers() != null + && !currentFlow.flow().getContainers().isEmpty(); + } + + private Container resolveExistingContainerForPlan(AssistantContainerPlan containerPlan, + FlowCreateRequest currentFlow, int planIndex, Set usedExistingContainerIds) { + List> existingContainers = currentFlow == null || currentFlow.flow() == null + || currentFlow.flow().getContainers() == null + ? List.of() + : currentFlow.flow().getContainers(); + if (existingContainers.isEmpty()) { + return null; + } + + Container exact = findExistingContainer(existingContainers, usedExistingContainerIds, + container -> Objects.equals(container.getId(), containerPlan.containerId()) + || normalizeBlockReference(containerPlan.containerId()) != null + && normalizeBlockReference(containerPlan.containerId()) + .equals(normalizeBlockReference(container.getName()))); + if (exact != null) { + return exact; + } + + Container byPurpose = findExistingContainer(existingContainers, usedExistingContainerIds, + container -> normalizeBlockReference(containerPlan.purpose()) != null + && normalizeBlockReference(containerPlan.purpose()) + .equals(normalizeBlockReference(container.getName()))); + if (byPurpose != null) { + return byPurpose; + } + + if (planIndex >= 0 && planIndex < existingContainers.size()) { + Container byPosition = existingContainers.get(planIndex); + if (!usedExistingContainerIds.contains(byPosition.getId())) { + return byPosition; + } + } + + List> unused = existingContainers.stream() + .filter(container -> !usedExistingContainerIds.contains(container.getId())) + .toList(); + return unused.size() == 1 ? unused.getFirst() : null; + } + + private Container findExistingContainer(List> existingContainers, + Set usedExistingContainerIds, java.util.function.Predicate> predicate) { + for (Container container : existingContainers) { + if (container == null || usedExistingContainerIds.contains(container.getId())) { + continue; + } + if (predicate.test(container)) { + return container; + } + } + return null; + } + private boolean canUseMinimalDraftFallback(String userPrompt, FlowCreateRequest currentFlow, Set availableBlockTypes) { return userPrompt != null 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 f4c08bd..7483473 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 @@ -48,6 +48,7 @@ import it.cnr.isti.workflow.manager.blocks.types.EndBlockType; 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.containers.Container; 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; @@ -1776,6 +1777,137 @@ public class AssistantControllerTest { .thenAnswer(answer); } + @Test + public void draftGroupsBlocksIntoGenericContainerAndWiresItAsOneNode() { + mockDraftResponsesWithGenericContainer(); + + AssistantFlowResponse response = assistantController.draft( + new AssistantGenerationRequest( + "summarize a document, then extract key facts and draft a review from the summary", + MODEL, + 1)); + + assertNotNull(response); + assertTrue(response.valid(), () -> "Unexpected validation errors: " + response.validationErrors()); + assertTrue(response.validationErrors().isEmpty()); + assertEquals(1, response.flow().flow().getBlocks().size()); + assertEquals(1, response.flow().flow().getContainers().size()); + + Container container = response.flow().flow().getContainers().getFirst(); + assertEquals("GenericContainer", container.getType().getName()); + assertEquals(2, container.getSpecificConfiguration().getSubFlow().getBlocks().size()); + assertEquals(1, container.getSpecificConfiguration().getSubFlow().getConnections().size()); + + // The container's exposed inputs/outputs are auto-derived from whichever inner handles + // are left unconnected - here, the first inner block's "summary" input and the second + // inner block's "response" output. + assertEquals(1, container.getInputs().size()); + assertEquals("summary", container.getInputs().getFirst().getName()); + assertEquals(1, container.getOutputs().size()); + assertEquals("response", container.getOutputs().getFirst().getName()); + + // Exactly one top-level connection: the standalone block feeding the container's + // exposed input. Nothing reaches into the container's own inner blocks from outside. + assertEquals(1, response.flow().flow().getConnections().size()); + Connection topLevelConnection = response.flow().flow().getConnections().getFirst(); + Block classifierBlock = response.flow().flow().getBlocks().getFirst(); + assertEquals(classifierBlock.getId(), topLevelConnection.getSourceId()); + assertEquals(container.getId(), topLevelConnection.getTargetId()); + assertEquals("summary", topLevelConnection.getTargetName()); + } + + private void mockDraftResponsesWithGenericContainer() { + Answer answer = invocation -> { + String prompt = invocation.getArgument(1, String.class); + if (prompt.contains("TASK: PLAN")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Summarize standalone, then group fact extraction and review drafting together.", + "plan", java.util.Map.of( + "name", "Document review pipeline", + "description", "Summarize a document, then extract facts and draft a review.", + "blocks", java.util.List.of( + java.util.Map.of( + "blockId", "b1", + "blockType", "LLMBlock", + "purpose", "Summarize the document")), + "containers", java.util.List.of( + java.util.Map.of( + "containerId", "c1", + "containerType", "GenericContainer", + "purpose", "Review pipeline", + "operation", "ADD", + "blocks", java.util.List.of( + java.util.Map.of( + "blockId", "c1-b1", + "blockType", "LLMBlock", + "purpose", "Extract key facts"), + java.util.Map.of( + "blockId", "c1-b2", + "blockType", "LLMBlock", + "purpose", "Draft the review"))))))); + } + 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 summarizer.", + "block", java.util.Map.of( + "blockId", "b1", + "name", "summarize-document", + "config", java.util.Map.of( + "prompt", "Summarize the following document: ${{document}}")))); + } + if (prompt.contains("TASK: BLOCK_CONFIG") + && prompt.contains("Current block to configure:\n{\n \"blockId\" : \"c1-b1\"")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Configured the fact extractor inside the container.", + "block", java.util.Map.of( + "blockId", "c1-b1", + "name", "extract-facts", + "config", java.util.Map.of( + "prompt", "Extract key facts from: ${{summary}}")))); + } + if (prompt.contains("TASK: BLOCK_CONFIG") + && prompt.contains("Current block to configure:\n{\n \"blockId\" : \"c1-b2\"")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Configured the review drafter inside the container.", + "block", java.util.Map.of( + "blockId", "c1-b2", + "name", "draft-review", + "config", java.util.Map.of( + "prompt", "Draft a review based on: ${{facts}}")))); + } + if (prompt.contains("TASK: CONNECTIONS") && !prompt.contains("\"blockId\" : \"c1\"")) { + // The inner, container-scoped connections call's configured-blocks summary only + // ever lists c1-b1/c1-b2 - the container's own id "c1" only appears in the + // top-level call's configured-blocks entry for the container itself. + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Connected fact extraction to review drafting inside the container.", + "connections", java.util.List.of( + java.util.Map.of( + "fromBlockId", "c1-b1", + "fromOutput", "response", + "toBlockId", "c1-b2", + "toInput", "facts")))); + } + if (prompt.contains("TASK: CONNECTIONS")) { + return TestAssistantResponses.wrap(java.util.Map.of( + "rationale", "Connected the summarizer to the container's exposed input.", + "connections", java.util.List.of( + java.util.Map.of( + "fromBlockId", "b1", + "fromOutput", "response", + "toBlockId", "c1", + "toInput", "summary")))); + } + 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); + } + static class TestAssistantResponses { private static final String PROVIDER = "InternalOllama";