From 6333623ee6535e344d50d998b1eac13b9768ec46 Mon Sep 17 00:00:00 2001 From: XianReallyHot-ZZH <741788769@qq.com> Date: Wed, 28 Oct 2026 12:03:29 +0800 Subject: [PATCH 1/2] fix(agent): deliver repaired structured output message as final ReAct output Adopt the finalizer's repaired message as the final delivered message in ReActAgent.finish: append it after the failed original (kept in history), notify stream consumers through AgentEvent.Message before Finished, and keep message, structured data and events consistent. Fixes #278. --- .../agentic/ai/agent/react/ReActAgent.java | 10 + .../ReActAgentStructuredDeliveryTest.java | 295 ++++++++++++++++++ 2 files changed, 305 insertions(+) create mode 100644 argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/react/ReActAgentStructuredDeliveryTest.java diff --git a/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/react/ReActAgent.java b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/react/ReActAgent.java index 6d88a5aa0..b34d52b6b 100644 --- a/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/react/ReActAgent.java +++ b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/react/ReActAgent.java @@ -591,6 +591,16 @@ private void finish(List messages, Map state, Usage if (result.status() != AgentResult.Status.COMPLETED) { throw new AgentExecutionException(result); } + // A successful repair replaces the delivered terminal: the message that + // actually passed validation becomes the final message while the failed + // original stays in history, and stream consumers see it through the + // complete-message event before the finished snapshot. + ChatMessage delivered = result.message(); + if (delivered != null && !delivered.text().equals(terminal.text()) + && !messages.isEmpty() && messages.get(messages.size() - 1) == terminal) { + messages.add(delivered); + emit(new AgentEvent.Message("model-repair", delivered)); + } if (result.data() != null) { outputs.put("structuredOutput", result.data()); } diff --git a/argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/react/ReActAgentStructuredDeliveryTest.java b/argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/react/ReActAgentStructuredDeliveryTest.java new file mode 100644 index 000000000..744406ad6 --- /dev/null +++ b/argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/react/ReActAgentStructuredDeliveryTest.java @@ -0,0 +1,295 @@ +/* + * Copyright 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.agent.react; + +import io.github.agentic.ai.agent.AgentExecutionContext; +import io.github.agentic.ai.agent.AgentEvent; +import io.github.agentic.ai.agent.AgentOutcome; +import io.github.agentic.ai.agent.AgentRequest; +import io.github.agentic.ai.agent.AgentResult; +import io.github.agentic.ai.api.chat.ChatClient; +import io.github.agentic.ai.api.chat.ChatClientCapabilities; +import io.github.agentic.ai.api.chat.ChatEvent; +import io.github.agentic.ai.api.chat.ChatMessage; +import io.github.agentic.ai.api.chat.ChatRequest; +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.agent.result.StructuredOutputFinalizer; +import io.github.agentic.ai.runtime.execution.DelegationScope; +import io.github.agentic.ai.runtime.execution.ExecutionBudget; +import io.github.agentic.ai.runtime.execution.PrincipalReference; +import io.github.agentic.ai.runtime.execution.TenantReference; +import io.github.agentic.ai.runtime.execution.TrustedExecutionContext; + +import java.time.Clock; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CompletionStage; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Flow; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Delivery consistency of the structured output boundary in the independent ReAct + * agent: whatever message actually passed validation is the message finally + * delivered, for the outcome snapshot and for stream consumers alike. + */ +class ReActAgentStructuredDeliveryTest { + + private static final String SCHEMA = "{\"type\":\"object\",\"required\":[\"answer\"]," + + "\"additionalProperties\":false,\"properties\":{\"answer\":{\"type\":\"string\"}}}"; + + private static final String REPAIRED = "{\"answer\":\"fixed\"}"; + + @Test + void validOriginalOutputIsDeliveredUnchanged() throws Exception { + AtomicInteger repairs = new AtomicInteger(); + ReActAgent agent = ReActAgent.builder(new ScriptedClient("{\"answer\":\"direct\"}")) + .outputSchema(SCHEMA) + .outputFinalizer(new StructuredOutputFinalizer(failingRepairClient(repairs))) + .build(); + + AgentOutcome outcome = agent.invoke(request(), context()).get(3, TimeUnit.SECONDS); + + assertEquals(AgentOutcome.Status.COMPLETED, outcome.status()); + assertEquals("{\"answer\":\"direct\"}", outcome.messages().get(outcome.messages().size() - 1).text()); + assertEquals("{\"answer\":\"direct\"}", outcome.outputs().get("structuredOutput").toString()); + assertEquals(0, repairs.get()); + } + + @Test + void successfulRepairDeliversRepairedMessageAsFinal() throws Exception { + ReActAgent agent = ReActAgent.builder(new ScriptedClient("not-json")) + .outputSchema(SCHEMA) + .outputFinalizer(new StructuredOutputFinalizer(scriptedRepairClient(REPAIRED))) + .build(); + + AgentOutcome outcome = agent.invoke(request(), context()).get(3, TimeUnit.SECONDS); + + assertEquals(AgentOutcome.Status.COMPLETED, outcome.status()); + // The repaired message is the final delivered message, agreeing with the data. + assertEquals(REPAIRED, outcome.messages().get(outcome.messages().size() - 1).text()); + assertEquals(REPAIRED, outcome.outputs().get("structuredOutput").toString()); + // The failed original stays in history right before it, not silently dropped. + assertEquals("not-json", outcome.messages().get(outcome.messages().size() - 2).text()); + } + + @Test + void repairStillInvalidKeepsFailureStateDistinct() throws Exception { + ReActAgent agent = ReActAgent.builder(new ScriptedClient("not-json")) + .outputSchema(SCHEMA) + .outputFinalizer(new StructuredOutputFinalizer(scriptedRepairClient("{\"wrong\":\"shape\"}"))) + .build(); + + ExecutionException failure = assertThrows(ExecutionException.class, + () -> agent.invoke(request(), context()).get(3, TimeUnit.SECONDS)); + + assertTrue(failure.getCause() instanceof AgentExecutionException); + AgentExecutionException delivery = (AgentExecutionException) failure.getCause(); + assertEquals(AgentResult.Status.VALIDATION_FAILED, delivery.result().status()); + assertTrue(delivery.result().issues().stream().anyMatch( + issue -> AgentResult.ERROR_CODE_REPAIRED_STILL_INVALID.equals(issue.errorCode()))); + } + + @Test + void repairClientFailureMarksRepairCallFailed() throws Exception { + ReActAgent agent = ReActAgent.builder(new ScriptedClient("not-json")) + .outputSchema(SCHEMA) + .outputFinalizer(new StructuredOutputFinalizer(brokenRepairClient())) + .build(); + + ExecutionException failure = assertThrows(ExecutionException.class, + () -> agent.invoke(request(), context()).get(3, TimeUnit.SECONDS)); + + assertTrue(failure.getCause() instanceof AgentExecutionException); + AgentExecutionException delivery = (AgentExecutionException) failure.getCause(); + assertEquals(AgentResult.Status.VALIDATION_FAILED, delivery.result().status()); + assertTrue(delivery.result().issues().stream().anyMatch( + issue -> AgentResult.ERROR_CODE_REPAIR_CALL_FAILED.equals(issue.errorCode()))); + } + + @Test + void streamingCompletionExposesRepairedFinalMessage() throws Exception { + ReActAgent agent = ReActAgent.builder(new StreamingClient("not-json")) + .streamModel(true) + .outputSchema(SCHEMA) + .outputFinalizer(new StructuredOutputFinalizer(scriptedRepairClient(REPAIRED))) + .build(); + List events = new ArrayList<>(); + CompletableFuture finished = new CompletableFuture<>(); + try (AgentExecutionContext context = context()) { + agent.stream(request(), context).subscribe(new Flow.Subscriber<>() { + public void onSubscribe(Flow.Subscription subscription) { + subscription.request(Long.MAX_VALUE); + } + + public void onNext(AgentEvent event) { + events.add(event); + if (event instanceof AgentEvent.Finished done) { + finished.complete(done.outcome()); + } + } + + public void onError(Throwable error) { + finished.completeExceptionally(error); + } + + public void onComplete() { + } + }); + AgentOutcome outcome = finished.get(3, TimeUnit.SECONDS); + assertEquals(REPAIRED, outcome.messages().get(outcome.messages().size() - 1).text()); + assertEquals(REPAIRED, outcome.outputs().get("structuredOutput").toString()); + } + // The repaired message reaches stream consumers before the finished snapshot, + // through the existing complete-message event contract. + int repairIndex = -1; + int finishIndex = -1; + for (int index = 0; index < events.size(); index++) { + AgentEvent event = events.get(index); + if (event instanceof AgentEvent.Message message && "model-repair".equals(message.messageId())) { + repairIndex = index; + assertEquals(REPAIRED, message.message().text()); + } + if (event instanceof AgentEvent.Finished) { + finishIndex = index; + } + } + assertTrue(repairIndex >= 0, "stream consumers must see the repaired final message"); + assertTrue(finishIndex > repairIndex, "the repaired message must precede the finished event"); + } + + private static AgentRequest request() { + return AgentRequest.messages(List.of(ChatMessage.user("hello"))); + } + + private static AgentExecutionContext context() { + TrustedExecutionContext trusted = TrustedExecutionContext.builder() + .tenant(new TenantReference("test", "tenant")) + .principal(new PrincipalReference("test", "principal")) + .runId("run").correlationId("correlation").budget(ExecutionBudget.unlimited()) + .delegationScope(DelegationScope.none()).build(); + return new AgentExecutionContext(trusted, Clock.systemUTC()); + } + + private static ChatResponse response(ChatMessage message) { + return new ChatResponse(null, "test", List.of(message), List.of(), FinishReason.STOP, new Usage(1L, 1L, 1L), + Map.of()); + } + + private static ChatClient scriptedRepairClient(String json) { + return new ChatClient() { + public CompletionStage call(ChatRequest request) { + return CompletableFuture.completedFuture(response(ChatMessage.assistant(json))); + } + + public Flow.Publisher stream(ChatRequest request) { + throw new UnsupportedOperationException(); + } + + public ChatClientCapabilities capabilities() { + return ChatClientCapabilities.basic(); + } + }; + } + + private static ChatClient brokenRepairClient() { + return new ChatClient() { + public CompletionStage call(ChatRequest request) { + return CompletableFuture.failedFuture(new IllegalStateException("repair backend down")); + } + + public Flow.Publisher stream(ChatRequest request) { + throw new UnsupportedOperationException(); + } + + public ChatClientCapabilities capabilities() { + return ChatClientCapabilities.basic(); + } + }; + } + + private static ChatClient failingRepairClient(AtomicInteger calls) { + return new ChatClient() { + public CompletionStage call(ChatRequest request) { + calls.incrementAndGet(); + throw new AssertionError("the repair client must not be called"); + } + + public Flow.Publisher stream(ChatRequest request) { + throw new UnsupportedOperationException(); + } + + public ChatClientCapabilities capabilities() { + return ChatClientCapabilities.basic(); + } + }; + } + + private static class ScriptedClient implements ChatClient { + + private final ChatResponse reply; + + ScriptedClient(String terminalText) { + this.reply = response(ChatMessage.assistant(terminalText)); + } + + public CompletionStage call(ChatRequest request) { + return CompletableFuture.completedFuture(reply); + } + + public Flow.Publisher stream(ChatRequest request) { + throw new UnsupportedOperationException(); + } + + public ChatClientCapabilities capabilities() { + return ChatClientCapabilities.basic(); + } + + } + + private static final class StreamingClient extends ScriptedClient { + + StreamingClient(String terminalText) { + super(terminalText); + } + + public Flow.Publisher stream(ChatRequest request) { + return subscriber -> subscriber.onSubscribe(new Flow.Subscription() { + public void request(long demand) { + subscriber.onNext(ChatEvent.textDelta("not-json")); + subscriber.onNext(ChatEvent.response(response(ChatMessage.assistant("not-json")))); + subscriber.onComplete(); + } + + public void cancel() { + } + }); + } + + } + +} From 7631a5a60b6862d90769f5ae4987c3d0a1583c0a Mon Sep 17 00:00:00 2001 From: XianReallyHot-ZZH <741788769@qq.com> Date: Wed, 28 Oct 2026 12:52:30 +0800 Subject: [PATCH 2/2] fix(agent): harden repair delivery guard and cover no-repair validation Compare message texts null-safely and drop the positional last-element guard so the repaired message is delivered on every termination path, and pin the no-repair-client validation-failure state with a test. --- .../agentic/ai/agent/react/ReActAgent.java | 3 +-- .../react/ReActAgentStructuredDeliveryTest.java | 17 +++++++++++++++++ 2 files changed, 18 insertions(+), 2 deletions(-) diff --git a/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/react/ReActAgent.java b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/react/ReActAgent.java index b34d52b6b..6d24699f5 100644 --- a/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/react/ReActAgent.java +++ b/argi-agent/argi-agent-core/src/main/java/io/github/agentic/ai/agent/react/ReActAgent.java @@ -596,8 +596,7 @@ private void finish(List messages, Map state, Usage // original stays in history, and stream consumers see it through the // complete-message event before the finished snapshot. ChatMessage delivered = result.message(); - if (delivered != null && !delivered.text().equals(terminal.text()) - && !messages.isEmpty() && messages.get(messages.size() - 1) == terminal) { + if (delivered != null && !Objects.equals(delivered.text(), terminal.text())) { messages.add(delivered); emit(new AgentEvent.Message("model-repair", delivered)); } diff --git a/argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/react/ReActAgentStructuredDeliveryTest.java b/argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/react/ReActAgentStructuredDeliveryTest.java index 744406ad6..9ea540587 100644 --- a/argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/react/ReActAgentStructuredDeliveryTest.java +++ b/argi-agent/argi-agent-core/src/test/java/io/github/agentic/ai/agent/react/ReActAgentStructuredDeliveryTest.java @@ -97,6 +97,23 @@ void successfulRepairDeliversRepairedMessageAsFinal() throws Exception { assertEquals("not-json", outcome.messages().get(outcome.messages().size() - 2).text()); } + @Test + void noRepairClientKeepsValidationFailedState() throws Exception { + ReActAgent agent = ReActAgent.builder(new ScriptedClient("not-json")) + .outputSchema(SCHEMA) + .outputFinalizer(new StructuredOutputFinalizer()) + .build(); + + ExecutionException failure = assertThrows(ExecutionException.class, + () -> agent.invoke(request(), context()).get(3, TimeUnit.SECONDS)); + + assertTrue(failure.getCause() instanceof AgentExecutionException); + AgentExecutionException delivery = (AgentExecutionException) failure.getCause(); + assertEquals(AgentResult.Status.VALIDATION_FAILED, delivery.result().status()); + assertTrue(delivery.result().issues().stream().anyMatch( + issue -> AgentResult.ERROR_CODE_INVALID_JSON.equals(issue.errorCode()))); + } + @Test void repairStillInvalidKeepsFailureStateDistinct() throws Exception { ReActAgent agent = ReActAgent.builder(new ScriptedClient("not-json"))