fix(agent): preserve flow history and isolate stream results

argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/ReactAgent.java

@@ -71,6 +71,7 @@
 import io.github.agentic.ai.graph.streaming.StreamingOutput;
 import io.github.agentic.ai.graph.serializer.StateSerializer;
 import io.github.agentic.ai.graph.serializer.plain_text.jackson.ArgiJacksonStateSerializer;
+import io.github.agentic.ai.graph.state.ReplaceAllWith;
 import io.github.agentic.ai.graph.state.strategy.AppendStrategy;
 import io.github.agentic.ai.graph.state.strategy.ReplaceStrategy;
 import io.github.agentic.ai.graph.utils.TypeRef;
@@ -355,6 +356,11 @@ public AgentResult callResult(Map<String, Object> inputs, RunnableConfig config)
 	 * outcome of the terminal message (one of the four terminal statuses), finalized
 	 * through the same boundary as {@link #callResult(String)}. A run that ends without
 	 * a terminal response is reported as INTERRUPTED, never as completed.
+	 * <p>
+	 * Delivery aggregation is isolated per subscription. Subscribing again can execute
+	 * the underlying agent again; results are not cached and tool side effects are not
+	 * made idempotent by this method.
+	 * </p>
 	 * @param message the user input message
 	 * @return the provisional chunk stream terminated by the validated result
 	 */
@@ -448,37 +454,41 @@ private Flux<AgentResult> doStreamResult(Map<String, Object> inputs, RunnableCon
 	 * results (status {@code null}); the final aggregated response (marked by the
 	 * existing {@code argiFinalResponse} protocol convention) is buffered and finalized
 	 * once the stream completes, so unverified chunks are never reported as a delivery.
+	 * Aggregation is local to each subscription. Re-subscribing still subscribes to the
+	 * underlying execution again; it does not cache results or make side effects idempotent.
 	 */
 	private Flux<AgentResult> finalizeStream(Flux<NodeOutput> outputs) {
-		AtomicReference<ChatResponse> finalResponse = new AtomicReference<>();
-		return outputs.concatMap(output -> {
-			if (!(output instanceof StreamingOutput<?> streamed)) {
-				return Flux.empty();
-			}
-			if (streamed.getOriginData() instanceof ChatResponse response
-					&& Boolean.TRUE.equals(response.metadata().get(FINAL_RESPONSE_METADATA_KEY))) {
-				finalResponse.set(response);
+		return Flux.defer(() -> {
+			AtomicReference<ChatResponse> finalResponse = new AtomicReference<>();
+			return outputs.concatMap(output -> {
+				if (!(output instanceof StreamingOutput<?> streamed)) {
+					return Flux.empty();
+				}
+				if (streamed.getOriginData() instanceof ChatResponse response
+						&& Boolean.TRUE.equals(response.metadata().get(FINAL_RESPONSE_METADATA_KEY))) {
+					finalResponse.set(response);
+					return Flux.empty();
+				}
+				if (streamed.getOutputType() == OutputType.AGENT_MODEL_STREAMING) {
+					return Flux.fromIterable(streamed.messages())
+						.map(message -> new AgentResult(message, null, Usage.unknown(), null, List.of()));
+				}
 				return Flux.empty();
-			}
-			if (streamed.getOutputType() == OutputType.AGENT_MODEL_STREAMING) {
-				return Flux.fromIterable(streamed.messages())
-					.map(message -> new AgentResult(message, null, Usage.unknown(), null, List.of()));
-			}
-			return Flux.empty();
-		}).concatWith(Flux.defer(() -> {
-			ChatResponse response = finalResponse.get();
-			if (response == null) {
-				return Flux.just(AgentResult.interrupted(Usage.unknown()));
-			}
-			ChatMessage terminal = response.messages().stream()
-				.filter(message -> message.role() == ChatRole.ASSISTANT)
-				.reduce((first, second) -> second)
-				.orElse(null);
-			if (terminal == null) {
-				return Flux.just(AgentResult.interrupted(response.usage()));
-			}
-			return Flux.just(deliver(terminal, response.usage()));
-		}));
+			}).concatWith(Flux.defer(() -> {
+				ChatResponse response = finalResponse.get();
+				if (response == null) {
+					return Flux.just(AgentResult.interrupted(Usage.unknown()));
+				}
+				ChatMessage terminal = response.messages().stream()
+					.filter(message -> message.role() == ChatRole.ASSISTANT)
+					.reduce((first, second) -> second)
+					.orElse(null);
+				if (terminal == null) {
+					return Flux.just(AgentResult.interrupted(response.usage()));
+				}
+				return Flux.just(deliver(terminal, response.usage()));
+			}));
+		});
 	}
 
 	private AgentResult doResultInvoke(Object message, RunnableConfig config) {
@@ -1401,8 +1411,11 @@ public Map<String, Object> apply(OverAllState parentState, RunnableConfig config
 				else {
 					newMessages = new ArrayList<>();
 				}
-				stateForChild.put("messages", canonicalizeMessages(newMessages));
-				subGraphResult = childGraph.graphResponseStream(stateForChild, subGraphRunnableConfig);
+				stateForChild.put("messages", ReplaceAllWith.of(canonicalizeMessages(newMessages)));
+				// Apply the snapshot marker before state serialization so message types stay intact.
+				OverAllState childInput = new OverAllState(stateForChild, childGraph.compileConfig.getStore());
+				childInput.registerKeyAndStrategy(childGraph.getKeyStrategyMap());
+				subGraphResult = childGraph.graphResponseStream(childInput, subGraphRunnableConfig);
 			}
 			else {
 				Map<String, Object> stateForChild = childStateWithoutAuthority(parentState.data());
@@ -1420,31 +1433,34 @@ public Map<String, Object> apply(OverAllState parentState, RunnableConfig config
 
 		private @NonNull Flux<GraphResponse<NodeOutput>> getGraphResponseFlux(OverAllState parentState,
 				Flux<GraphResponse<NodeOutput>> subGraphResult, AgentInstructionMessage instructionMessage) {
-			// Use buffer(2, 1) to create sliding windows: [elem0, elem1], [elem1, elem2],
-			// ..., [elemN-1, elemN], [elemN]
-			// For windows with 2 elements, emit the first (previous element)
-			// For the last window with 1 element, process it specially
-			return subGraphResult.buffer(2, 1).flatMap(window -> {
-				if (window.size() == 1) {
-					// Last window: process the last element with message filtering
-					return Flux.just(processLastResponse(window.get(0), parentState, instructionMessage));
-				}
-				else {
-					// Regular window: emit the first element (previous, delayed by one)
-					return Flux.just(window.get(0));
-				}
-			}, 1); // Concurrency of 1 to maintain order
+			return Flux.defer(() -> {
+				AtomicReference<List<?>> inheritedMessages = new AtomicReference<>(includeContents
+						? canonicalizeMessages(new ArrayList<>(parentState.<List<?>>value("messages").orElse(List.of())))
+						: List.of());
+				return subGraphResult.doOnNext(response -> {
+					if (response.getOutput() != null && !response.getOutput().isCompletedExceptionally()) {
+						NodeOutput output = response.getOutput().join();
+						if (START.equals(output.node()) && !output.isSubGraph()) {
+							// Capture the child's actual initial history, including restored checkpoint messages.
+							inheritedMessages.set(new ArrayList<>(output.state().<List<?>>value("messages")
+								.orElse(List.of())));
+						}
+					}
+				}).buffer(2, 1).map(window -> window.size() == 1
+						? processLastResponse(window.get(0), inheritedMessages.get(), instructionMessage)
+						: window.get(0));
+			});
 		}
 
 		/**
 		 * Process the last response by filtering messages based on parent state and
 		 * returnReasoningContents flag.
 		 * @param lastResponse the last response from sub-graph
-		 * @param parentState the parent state containing messages to filter out
+		 * @param inheritedMessages the initial child history whose occurrences are already owned by the parent
 		 * @return processed GraphResponse with filtered messages
 		 */
 		private GraphResponse<NodeOutput> processLastResponse(GraphResponse<NodeOutput> lastResponse,
-				OverAllState parentState, AgentInstructionMessage instructionMessage) {
+				List<?> inheritedMessages, AgentInstructionMessage instructionMessage) {
 			if (lastResponse == null) {
 				return lastResponse;
 			}
@@ -1458,11 +1474,10 @@ private GraphResponse<NodeOutput> processLastResponse(GraphResponse<NodeOutput>
 						@SuppressWarnings("unchecked")
 						List<Object> messages = new ArrayList<>((List<Object>) resultMap.get("messages"));
 						if (!messages.isEmpty()) {
-							parentState.value("messages").ifPresent(parentMsgs -> {
-								if (parentMsgs instanceof List) {
-									messages.removeAll((List<?>) parentMsgs);
-								}
-							});
+							// Subtract each inherited occurrence once; removeAll would also erase new equal messages.
+							for (Object inherited : inheritedMessages) {
+								messages.remove(inherited);
+							}
 
 							List<Object> finalMessages;
 							if (returnReasoningContents) {
@@ -1486,7 +1501,7 @@ private GraphResponse<NodeOutput> processLastResponse(GraphResponse<NodeOutput>
 
 							Map<String, Object> newResultMap = new HashMap<>(resultMap);
 							newResultMap.put("messages", tagChildHandoffMessages(finalMessages));
-							return GraphResponse.done(newResultMap);
+							return GraphResponse.done(newResultMap, lastResponse.getAllMetadata());
 						}
 					}
 				}

argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/flow/agent/FlowAgent.java

@@ -29,6 +29,13 @@
 import java.util.List;
 import java.util.concurrent.Executor;
 
+/**
+ * Composes agents while preserving each conversation message occurrence. Repeated user
+ * inputs and equal child answers are appended as separate events. A child receives a
+ * conversation snapshot and returns only its newly produced message occurrences, so
+ * inherited history is not appended again. This behavior also applies after checkpoint
+ * restoration; existing checkpoints require no format migration.
+ */
 public abstract class FlowAgent extends Agent {
 
 	protected List<String> interruptBefore;

argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/flow/strategy/FlowGraphBuildingStrategy.java

@@ -24,7 +24,6 @@
 import io.github.agentic.ai.graph.agent.flow.agent.FlowAgent;
 import io.github.agentic.ai.graph.agent.flow.builder.FlowGraphBuilder;
 import io.github.agentic.ai.graph.exception.GraphStateException;
-import io.github.agentic.ai.graph.internal.node.SubCompiledGraphNode;
 import io.github.agentic.ai.graph.state.strategy.AppendStrategy;
 import io.github.agentic.ai.graph.state.strategy.ReplaceStrategy;
 
@@ -97,7 +96,8 @@ default KeyStrategyFactory generateKeyStrategyFactory(FlowGraphBuilder.FlowGraph
 				}
 			}
 
-			keyStrategyMap.put("messages", new AppendStrategy(false));
+			// Each invocation and child delta represents new occurrences, even when text repeats.
+			keyStrategyMap.put("messages", new AppendStrategy());
 
 			return keyStrategyMap;
 		};
@@ -145,7 +145,7 @@ default void processOutputKey(String outputKey, KeyStrategy outputKeyStrategy, M
 
 	static void addSubAgentNode(Agent subAgent, StateGraph newGraph) throws GraphStateException {
 		if (subAgent instanceof FlowAgent flowAgent) {
-			newGraph.addNode(flowAgent.name(), new SubCompiledGraphNode(flowAgent.name(), flowAgent.getAndCompileGraph()));
+			newGraph.addNode(flowAgent.name(), new FlowSubGraphNode(flowAgent.name(), flowAgent.getAndCompileGraph()));
 		} else if (subAgent instanceof BaseAgent baseAgent) {
 			newGraph.addNode(baseAgent.name(), baseAgent.asNode(baseAgent.isIncludeContents(), baseAgent.isReturnReasoningContents()));
 		} else {

argi-agent-framework/src/main/java/io/github/agentic/ai/graph/agent/flow/strategy/FlowSubGraphNode.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2025-2026 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.github.agentic.ai.graph.agent.flow.strategy;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+
+import io.github.agentic.ai.graph.CompiledGraph;
+import io.github.agentic.ai.graph.OverAllState;
+import io.github.agentic.ai.graph.RunnableConfig;
+import io.github.agentic.ai.graph.action.AsyncNodeActionWithConfig;
+import io.github.agentic.ai.graph.internal.node.ResumableSubGraphAction;
+import io.github.agentic.ai.graph.internal.node.SubCompiledGraphNode;
+import io.github.agentic.ai.graph.internal.node.SubCompiledGraphNodeAction;
+import io.github.agentic.ai.graph.state.ReplaceAllWith;
+
+/** Passes the parent's conversation snapshot to a nested flow without appending it twice. */
+final class FlowSubGraphNode extends SubCompiledGraphNode {
+
+	private final CompiledGraph flowGraph;
+
+	FlowSubGraphNode(String id, CompiledGraph flowGraph) {
+		super(id, flowGraph);
+		this.flowGraph = flowGraph;
+	}
+
+	@Override
+	public ActionFactory actionFactory() {
+		return parentConfig -> new SnapshotAction(new SubCompiledGraphNodeAction(id(), parentConfig, flowGraph));
+	}
+
+	private record SnapshotAction(SubCompiledGraphNodeAction delegate)
+			implements AsyncNodeActionWithConfig, ResumableSubGraphAction {
+
+		@Override
+		public String getResumeSubGraphId() {
+			return delegate.getResumeSubGraphId();
+		}
+
+		@Override
+		public CompletableFuture<Map<String, Object>> apply(OverAllState state, RunnableConfig config) {
+			Map<String, Object> inputs = new HashMap<>(state.data());
+			if (inputs.get("messages") instanceof List<?> messages) {
+				inputs.put("messages", ReplaceAllWith.of(new ArrayList<>(messages)));
+			}
+			OverAllState child = new OverAllState(inputs, state.getStore());
+			child.registerKeyAndStrategy(state.keyStrategies());
+			return delegate.apply(child, config);
+		}
+
+	}
+
+}

argi-agent-framework/src/test/java/io/github/agentic/ai/graph/agent/ReactAgentStreamResultTest.java

@@ -30,10 +30,16 @@
 import io.github.agentic.ai.api.chat.ChatResponse;
 import io.github.agentic.ai.api.chat.FinishReason;
 import io.github.agentic.ai.api.chat.Usage;
+import io.github.agentic.ai.graph.NodeOutput;
+import io.github.agentic.ai.graph.OverAllState;
+import io.github.agentic.ai.graph.streaming.OutputType;
+import io.github.agentic.ai.graph.streaming.StreamingOutput;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 import reactor.adapter.JdkFlowAdapter;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Sinks;
+import org.springframework.test.util.ReflectionTestUtils;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -135,6 +141,82 @@ private static ChatResponse finalResponse(String text, Usage usage) {
 				FinishReason.STOP, usage, Map.of("argiFinalResponse", true));
 	}
 
+	@Test
+	void repeatedSubscriptionDoesNotReuseAnEarlierFinalResponse() {
+		AtomicInteger subscriptions = new AtomicInteger();
+		Flux<NodeOutput> source = Flux.defer(() -> subscriptions.getAndIncrement() == 0
+				? Flux.just(terminalOutput("first", 5L)) : Flux.empty());
+		Flux<AgentResult> results = finalizeOutputs(source);
+
+		assertThat(results.blockLast().message().text()).isEqualTo("first");
+		AgentResult interrupted = results.blockLast();
+		assertThat(interrupted.status()).isEqualTo(AgentResult.Status.INTERRUPTED);
+		assertThat(interrupted.message()).isNull();
+		assertThat(interrupted.usage()).isEqualTo(Usage.unknown());
+	}
+
+	@Test
+	void overlappingSubscriptionsKeepTheirOwnChunksFinalResponseAndUsage() {
+		Sinks.Many<NodeOutput> first = Sinks.many().unicast().onBackpressureBuffer();
+		Sinks.Many<NodeOutput> second = Sinks.many().unicast().onBackpressureBuffer();
+		AtomicInteger subscriptions = new AtomicInteger();
+		Flux<AgentResult> results = finalizeOutputs(Flux.defer(() ->
+			subscriptions.getAndIncrement() == 0 ? first.asFlux() : second.asFlux()));
+		CompletableFuture<List<AgentResult>> firstResults = results.collectList().toFuture();
+		CompletableFuture<List<AgentResult>> secondResults = results.collectList().toFuture();
+
+		first.emitNext(provisionalOutput("first chunk"), Sinks.EmitFailureHandler.FAIL_FAST);
+		first.emitNext(terminalOutput("first", 5L), Sinks.EmitFailureHandler.FAIL_FAST);
+		second.emitNext(provisionalOutput("second chunk"), Sinks.EmitFailureHandler.FAIL_FAST);
+		second.emitNext(terminalOutput("second", 9L), Sinks.EmitFailureHandler.FAIL_FAST);
+		first.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST);
+		second.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST);
+
+		assertThat(firstResults.join()).hasSize(2);
+		assertThat(firstResults.join().get(0).message().text()).isEqualTo("first chunk");
+		assertThat(firstResults.join().get(0).status()).isNull();
+		assertThat(firstResults.join().get(1).message().text()).isEqualTo("first");
+		assertThat(firstResults.join().get(1).usage().totalTokens()).isEqualTo(5L);
+		assertThat(secondResults.join()).hasSize(2);
+		assertThat(secondResults.join().get(0).message().text()).isEqualTo("second chunk");
+		assertThat(secondResults.join().get(1).message().text()).isEqualTo("second");
+		assertThat(secondResults.join().get(1).usage().totalTokens()).isEqualTo(9L);
+	}
+
+	@Test
+	void cancelledSubscriptionCannotSupplyAnotherSubscriptionsFinalResponse() {
+		Sinks.Many<NodeOutput> first = Sinks.many().unicast().onBackpressureBuffer();
+		Sinks.Many<NodeOutput> second = Sinks.many().unicast().onBackpressureBuffer();
+		AtomicInteger subscriptions = new AtomicInteger();
+		Flux<AgentResult> results = finalizeOutputs(Flux.defer(() ->
+			subscriptions.getAndIncrement() == 0 ? first.asFlux() : second.asFlux()));
+		CompletableFuture<List<AgentResult>> cancelled = results.collectList().toFuture();
+		CompletableFuture<List<AgentResult>> continuing = results.collectList().toFuture();
+		first.emitNext(terminalOutput("cancelled", 5L), Sinks.EmitFailureHandler.FAIL_FAST);
+		assertThat(cancelled.cancel(true)).isTrue();
+		second.emitNext(provisionalOutput("partial"), Sinks.EmitFailureHandler.FAIL_FAST);
+		second.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST);
+
+		assertThat(continuing.join()).hasSize(2);
+		assertThat(continuing.join().get(0).message().text()).isEqualTo("partial");
+		assertThat(continuing.join().get(1).status()).isEqualTo(AgentResult.Status.INTERRUPTED);
+		assertThat(continuing.join().get(1).message()).isNull();
+	}
+
+	private static Flux<AgentResult> finalizeOutputs(Flux<NodeOutput> source) {
+		return ReflectionTestUtils.invokeMethod(agent(new ScriptedStreamingClient(), null), "finalizeStream", source);
+	}
+
+	private static NodeOutput terminalOutput(String text, long tokens) {
+		return new StreamingOutput<>(finalResponse(text, new Usage(null, null, tokens)), "model",
+				"agent", new OverAllState(), OutputType.AGENT_MODEL_STREAMING);
+	}
+
+	private static NodeOutput provisionalOutput(String text) {
+		return new StreamingOutput<>(ChatMessage.assistant(text), "model", "agent",
+				new OverAllState(), OutputType.AGENT_MODEL_STREAMING);
+	}
+
 	private static ReactAgent agent(ChatClient client, String outputSchema) {
 		Builder builder = ReactAgent.builder()
 			.name("stream-result-test")