{"record":{"id":"942e977d8f2b83e7","repo":"alibaba/spring-ai-alibaba","slug":"node-stream-must-return-map-result-using-data-done","errorCode":null,"errorMessage":"Node stream must return Map result using Data.done(),","messagePattern":"Node stream must return Map result using Data\\.done\\(\\),","errorType":"validation","errorClass":"IllegalArgumentException","httpStatus":null,"severity":"error","filePath":"spring-ai-alibaba-graph-core/src/main/java/com/alibaba/cloud/ai/graph/executor/NodeExecutor.java","lineNumber":588,"sourceCode":"\t\t\tMap<String, Object> partialStateWithoutFlux = partialState.entrySet()\n\t\t\t\t\t.stream()\n\t\t\t\t\t.filter(e -> !(e.getValue() instanceof Flux) \n\t\t\t\t\t\t\t&& !(e.getValue() instanceof GraphFlux)\n\t\t\t\t\t\t\t&& !(e.getValue() instanceof ParallelGraphFlux))\n\t\t\t\t\t.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));\n\n\t\t\tMap<String, Object> updateState = new HashMap<>();\n\t\t\tif (nodeResultValue.isPresent()) {\n\t\t\t\tObject value = nodeResultValue.get();\n\t\t\t\tif (value instanceof Map<?, ?>) {\n\t\t\t\t\tupdateState = (Map<String, Object>) value;\n\t\t\t\t}\n\t\t\t\telse if (value instanceof BaseCheckpointSaver.Tag) {\n\t\t\t\t\t// When releaseThread=true, completion may return a Tag.\n\t\t\t\t\t// Tag is a checkpoint-release receipt, not a state update, so skip merge here.\n\t\t\t\t}\n\t\t\t\telse {\n\t\t\t\t\tthrow new IllegalArgumentException(\"Node stream must return Map result using Data.done(),\");\n\t\t\t\t}\n\t\t\t}\n\n\t\t\tMap<String, Object> combinedUpdateState = new HashMap<>(partialStateWithoutFlux);\n\t\t\tcombinedUpdateState.putAll(updateState);\n\t\t\tOptional<InterruptionMetadata> interruptAfterMetadata = interruptAfterForStreaming(context, combinedUpdateState);\n\n\t\t\tcontext.mergeIntoCurrentState(partialStateWithoutFlux);\n\t\t\tcontext.mergeIntoCurrentState(updateState);\n\n\t\t\ttry {\n\t\t\t\tCommand nextCommand = context.nextNodeId(context.getCurrentNodeId(), context.getCurrentStateData());\n\t\t\t\tcontext.setNextNodeId(nextCommand.gotoNode());\n\n\t\t\t\tcontext.buildNodeOutputAndAddCheckpoint(updateState);\n\n\t\t\t\tcontext.doListeners(NODE_AFTER, null);\n\t\t\t\tinterruptAfterMetadata.ifPresent(context::setReturnFromEmbedWithValue);","sourceCodeStart":570,"sourceCodeEnd":606,"githubUrl":"https://github.com/alibaba/spring-ai-alibaba/blob/f82da0b50f35744c13968191be2b1cd2452ef550/spring-ai-alibaba-graph-core/src/main/java/com/alibaba/cloud/ai/graph/executor/NodeExecutor.java#L570-L606","documentation":"In NodeExecutor.processGraphResponseFlux, when a node's stream completes, the final done value must be a Map (state update, via Data.done()), while certain special values (InterruptionMetadata, BaseCheckpointSaver.Tag) are handled explicitly. Anything else triggers IllegalArgumentException 'Node stream must return Map result using Data.done(),' telling the developer the stream's completion contract was violated.","triggerScenarios":"A node's embedded Flux completing with Data.done(nonMapValue), or emitting a final element that is neither Map, InterruptionMetadata, nor BaseCheckpointSaver.Tag — e.g. done with a String, POJO, or null wrapper.","commonSituations":"Custom streaming nodes returning domain objects instead of state maps at completion; misunderstanding that Data.done() must carry a Map<String,Object> state update; refactoring a node from sync (returning POJO) to streaming without wrapping the result in a Map.","solutions":["Complete the stream with Data.done(Map<String,Object>) — the map is merged into the graph state.","If the final value is interruption feedback, return InterruptionMetadata (handled natively); for checkpoint release receipts return BaseCheckpointSaver.Tag.","Wrap a POJO result in a state map before calling done: Data.done(Map.of(\"result\", pojo))."],"exampleFix":"// before\nreturn Flux.just(chunk).concatWith(Flux.just(Data.done(myPojo)));\n// after\nreturn Flux.just(chunk).concatWith(Flux.just(Data.done(Map.of(\"result\", myPojo))));","handlingStrategy":"validation","validationCode":"if (!(doneValue instanceof Map) && !(doneValue instanceof InterruptionMetadata) && !(doneValue instanceof BaseCheckpointSaver.Tag)) {\n    throw new IllegalStateException(\"Data.done() must carry a Map<String,Object> state update\");\n}","typeGuard":"boolean isValidDoneResult(Object o) {\n    return o instanceof Map<?, ?> || o instanceof InterruptionMetadata || o instanceof BaseCheckpointSaver.Tag;\n}","tryCatchPattern":"try {\n    return processGraphResponseFlux(response, context);\n} catch (IllegalArgumentException e) {\n    if (e.getMessage() != null && e.getMessage().contains(\"Data.done()\")) {\n        throw new IllegalStateException(\"Node stream completed with unsupported done value; wrap result in a Map\", e);\n    }\n    throw e;\n}","preventionTips":["Always finish node streams with Data.done(Map<String,Object>).","Wrap POJO results in Map.of(\"result\", pojo) rather than passing them raw.","Review the streaming-node contract docs when migrating sync nodes to streaming."],"tags":["streaming","flux","state-update","contract-violation"],"backgroundTag":"unsupported-operation","analyzedSha":"f82da0b50f35744c13968191be2b1cd2452ef550","analyzedAt":"2026-09-09T15:32:42.421Z","contentChangedAt":"2026-09-09T15:32:42.421Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}