{"record":{"id":"377f45b6d2db7148","repo":"alibaba/spring-ai-alibaba","slug":"failed-to-check-interruptafter-hook-for-streaming","errorCode":null,"errorMessage":"Failed to check interruptAfter hook for streaming node","messagePattern":"Failed to check interruptAfter hook for streaming node","errorType":"exception","errorClass":"RuntimeException","httpStatus":null,"severity":"error","filePath":"spring-ai-alibaba-graph-core/src/main/java/com/alibaba/cloud/ai/graph/executor/NodeExecutor.java","lineNumber":844,"sourceCode":"\t * @return interruption metadata if the hook triggers\n\t */\n\tprivate Optional<InterruptionMetadata> interruptAfterForStreaming(GraphRunnerContext context,\n\t\t\tMap<String, Object> actionResult) {\n\t\tString currentNodeId = context.getCurrentNodeId();\n\t\tAsyncNodeActionWithConfig action = context.getNodeAction(currentNodeId);\n\n\t\tif (!(action instanceof InterruptableAction interruptableAction)) {\n\t\t\treturn Optional.empty();\n\t\t}\n\n\t\ttry {\n\t\t\tOverAllState stateBeforeMerge = context.cloneState(context.getCurrentStateData());\n\t\t\treturn interruptableAction.interruptAfter(currentNodeId, stateBeforeMerge, actionResult,\n\t\t\t\t\tcontext.getConfig());\n\t\t}\n\t\tcatch (Exception e) {\n\t\t\tcontext.doListeners(ERROR, e);\n\t\t\tthrow new RuntimeException(\"Failed to check interruptAfter hook for streaming node\", e);\n\t\t}\n\t}\n\n\t/**\n\t * Handles ParallelGraphFlux processing with node ID preservation for all parallel streams.\n\t * @param context the graph runner context\n\t * @param parallelGraphFlux the ParallelGraphFlux to handle\n\t * @param partialState the partial state\n\t * @param resultValue the atomic reference to store the result value\n\t * @return Flux of GraphResponse with ParallelGraphFlux handling result\n\t */\n\tprivate Flux<GraphResponse<NodeOutput>> handleParallelGraphFlux(GraphRunnerContext context,\n\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\tParallelGraphFlux parallelGraphFlux, Map<String, Object> partialState,\n\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\t\tAtomicReference<Object> resultValue) throws Exception {\n\n\t\tif (parallelGraphFlux.isEmpty()) {\n\t\t\t// Handle empty ParallelGraphFlux\n\t\t\treturn handleNonStreamingResult(context, partialState, resultValue);","sourceCodeStart":826,"sourceCodeEnd":862,"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#L826-L862","documentation":"NodeExecutor wraps any exception thrown while evaluating the interruptAfter hook of a streaming node into a RuntimeException with this message. The interruptAfter mechanism lets a graph pause after a node completes so the caller can inspect or modify state; a failure here aborts node execution. The original exception is preserved as the cause.","triggerScenarios":"An exception is thrown by interruptableAction.interruptAfter(currentNodeId, stateBeforeMerge, actionResult, config) — e.g. a user-supplied InterruptAfterAction fails, or context.cloneState(context.getCurrentStateData()) throws while snapshotting state before the hook.","commonSituations":"Custom interrupt/human-in-the-loop hooks with bugs (NPEs on null state fields); state classes that are not properly cloneable/serializable in streaming mode; listener registered via doListeners throwing; concurrent state mutation during streaming execution.","solutions":["Inspect the cause chain (getCause()) to find the real exception from the interruptAfter hook or cloneState.","Fix the custom InterruptAfterAction implementation to handle null/empty state fields defensively.","Verify state values are cloneable/serializable if using checkpointed state with streaming nodes.","Check registered graph listeners for exceptions thrown during the ERROR/After-node lifecycle events.","Temporarily remove interruptAfter from the CompileConfig to confirm the hook is the failure point."],"exampleFix":"// before: hook assumes state value is always present\nreturn state.value(\"draft\").map(...);\n// after: guard against missing value\nObject draft = state.value(\"draft\").orElse(null);\nif (draft == null) { return Optional.empty(); }\nreturn Optional.of(new InterruptMetadata(nodeId, draft));","handlingStrategy":"try-catch","validationCode":"// Before compiling, sanity-check the hook compiles/returns\nInterruptAfterAction hook = ...;\ntry { hook.interruptAfter(nodeId, state, result, config); } catch (Exception e) { log.error(\"hook precheck failed\", e); }","typeGuard":"boolean safeState(OverAllState s) { return s != null && s.data() != null; }","tryCatchPattern":"try { graph.invoke(inputs, config); } catch (RuntimeException e) { if (e.getMessage() != null && e.getMessage().contains(\"interruptAfter hook\")) { Throwable root = e.getCause(); /* handle hook failure */ } }","preventionTips":["Keep interruptAfter hooks null-safe over state values","Ensure state is cloneable when using streaming + interrupts","Avoid throwing from graph listeners","Test human-in-the-loop hooks with checkpointing enabled in CI"],"tags":["streaming","human-in-the-loop","interrupt","graph-execution"],"backgroundTag":"internal-invariant-violation","analyzedSha":"f82da0b50f35744c13968191be2b1cd2452ef550","analyzedAt":"2026-09-09T15:32:42.421Z","contentChangedAt":"2026-09-09T15:32:42.421Z","schemaVersion":2},"datasetVersion":"2026-09-14T05:17:10.506Z"}