From f023b6685ab8219b1d7ead83817b0d47d4bceae4 Mon Sep 17 00:00:00 2001 From: matevip Date: Fri, 23 Oct 2026 04:29:00 -0400 Subject: [PATCH] fix(agent): retry finalization after reasoning exhausts output budget (#672) --- .../agent/graph/NodeStreamingChatHelper.java | 14 +- .../guard/TaskFinalizationCoordinator.java | 72 +++++++--- ...odeStreamingChatHelperThinkingCapTest.java | 15 ++ .../TaskFinalizationCoordinatorTest.java | 129 ++++++++++++++++++ .../ReasoningFinalizationTransportTest.java | 113 +++++++++++++++ 5 files changed, 321 insertions(+), 22 deletions(-) create mode 100644 mateclaw-server/src/test/java/vip/mate/llm/chatmodel/ReasoningFinalizationTransportTest.java diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java index ed7435cb..d5ac7042 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/NodeStreamingChatHelper.java @@ -1709,12 +1709,16 @@ public class NodeStreamingChatHelper { : truncatedByContentRepeat ? "content_repetition" : thinkingTokenLimit ? "thinking_token_limit" : null; + // Retain the provider's end condition even when it emitted some content. + // A verifier JSON cut off by length is not a malformed completed verdict. + Map resultMetadata = new LinkedHashMap<>(providerMetadata.get()); + if (finishReason.get() != null) resultMetadata.put("generationFinishReason", finishReason.get()); return assembleResult(contentAccum, thinkingAccum, toolCallAccumulators, promptTokens.get(), completionTokens.get(), cacheReadTokens.get(), cacheWriteTokens.get(), reasoningTokens.get(), phase, truncated, - truncationReason, providerMetadata.get()); + truncationReason, resultMetadata); } /** 组装 stopped partial 结果（用户主动停止，有已累积内容） */ @@ -2388,6 +2392,14 @@ public class NodeStreamingChatHelper { return errorType == ErrorType.PROMPT_TOO_LONG; } + /** Explicit provider output-limit stop, distinct from transport errors and loop guards. */ + public boolean isOutputTokenLimit() { + if (stopped || hasToolCalls || errorType != ErrorType.NONE) return false; + if ("thinking_token_limit".equals(errorMessage)) return true; + return errorMessage == null && assistantMessage != null + && "length".equalsIgnoreCase(String.valueOf(assistantMessage.getMetadata().get("generationFinishReason"))); + } + /** 是否有任何可保存的内容（text/thinking/toolCalls） */ public boolean hasAnyContent() { return (text != null && !text.isBlank()) diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/guard/TaskFinalizationCoordinator.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/guard/TaskFinalizationCoordinator.java index 5e9250a0..14213d65 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/guard/TaskFinalizationCoordinator.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/guard/TaskFinalizationCoordinator.java @@ -18,6 +18,7 @@ import java.util.concurrent.CancellationException; /** Call-scoped publication review. No conversation or attempt state is retained here. */ public final class TaskFinalizationCoordinator { private static final int OUTPUT_TOKENS = 1024; + private static final int MAX_RETRY_OUTPUT_TOKENS = 8192; private static final String RECOVERY_PROSE_RULES = """ Recovery prose must match the specific failed original request. Unqualified phrases such as "随后补齐 language="python" 重试", "补齐参数后成功", "fixed the parameters and retried", or @@ -95,7 +96,7 @@ public final class TaskFinalizationCoordinator { ACCEPTED, REVISION_REQUIRED, RECOVERY_RELATION_CORRECTION, CONTINUATION_REQUIRED, INVALID_VERDICT, FAILED_CALL_COUNT_MISMATCH, UNSUPPORTED_RECOVERY, CONTRADICTORY_OUTCOME, TASK_UNFINISHED, INVALID_CONTINUATION, CORRECTION_LIMIT, CONTINUATION_NOT_AVAILABLE, VERIFICATION_REJECTED, - CHECKER_UNAVAILABLE, CHECKER_PARTIAL, CHECKER_ERROR, CHECKER_TOOL_CALL, CHECKER_EXCEPTION, CANCELLED + CHECKER_UNAVAILABLE, CHECKER_PARTIAL, CHECKER_OUTPUT_LIMIT, CHECKER_ERROR, CHECKER_TOOL_CALL, CHECKER_EXCEPTION, CANCELLED } public record Review(TaskFinalizationGate.Result result, NodeStreamingChatHelper.StreamResult usage, int calls, boolean stopped) {} @@ -183,21 +184,46 @@ public final class TaskFinalizationCoordinator { "progressHintsTokens",includedProgressTokens,"progressHintsOmitted",progressOmitted, "canContinue",canContinue,"correctiveAttempts",attempts)); NodeStreamingChatHelper.StreamResult response=null; + NodeStreamingChatHelper.StreamResult usage=null; + int calls=0; String outcome="INCOMPLETE"; Diagnostic diagnostic=Diagnostic.CHECKER_UNAVAILABLE; TaskFinalizationGate.ParseFailureCode parseFailureCode=TaskFinalizationGate.ParseFailureCode.NOT_EVALUATED; try { var checkState = new vip.mate.agent.runtime.state.RunState(runState.data()); checkState.replace(vip.mate.agent.graph.state.MateClawStateKeys.REQUESTED_THINKING_LEVEL,"off"); - Prompt checkPrompt = new Prompt(messages,options); - response=vip.mate.agent.runtime.state.RuntimeExecutionScope.call(checkState, - () -> helper.streamCallSilent(model,checkPrompt,conversationId,"finalization_check")); - if(response==null) return incomplete("Checker unavailable",null,1,false); - if(stopped(conversationId) || response.stopped()) { diagnostic=Diagnostic.CANCELLED;return incomplete("Cancelled",response,1,true); } + for (int attempt=0; ; attempt++) { + Prompt checkPrompt = new Prompt(messages,options); + calls++; + response=vip.mate.agent.runtime.state.RuntimeExecutionScope.call(checkState, + () -> helper.streamCallSilent(model,checkPrompt,conversationId,"finalization_check")); + usage=usage==null?response:addUsage(usage,response); + if(response==null) return incomplete("Checker unavailable",usage,calls,false); + if(stopped(conversationId) || response.stopped()) { diagnostic=Diagnostic.CANCELLED;return incomplete("Cancelled",usage,calls,true); } + // Some compatible providers cannot disable reasoning. Their output + // limit includes thinking, so the small JSON allowance can expire + // before a complete verdict is emitted. Retry only that explicit condition, + // once, without replaying tools or dropping evidence from the prompt. + if(attempt==0 && response.isOutputTokenLimit() && callAvailable(maxIterations,usedCalls+calls)) { + int configured=vip.mate.agent.context.BudgetedCompactionModel.outputReserveTokens( + ChatModelDefaults.optionsFor(model),4096); + int available=window-TokenEstimator.estimateTokens(messages)-128; + int retryTokens=Math.min(Math.min(configured,MAX_RETRY_OUTPUT_TOKENS),available); + if(retryTokens>OUTPUT_TOKENS) { + options=toolFreeOptions(model,retryTokens); + continue; + } + } + break; + } + if(response.isOutputTokenLimit()) { + diagnostic=Diagnostic.CHECKER_OUTPUT_LIMIT; + return incomplete(diagnostic.name(),usage,calls,false); + } if(response.partial() || response.errorMessage()!=null || response.hasToolCalls()) { diagnostic=response.errorMessage()!=null?Diagnostic.CHECKER_ERROR :response.partial()?Diagnostic.CHECKER_PARTIAL:Diagnostic.CHECKER_TOOL_CALL; - return incomplete(diagnostic.name(),response,1,false); + return incomplete(diagnostic.name(),usage,calls,false); } var verdict=gate.parse(response.text()).orElse(null); parseFailureCode=verdict==null?gate.parseFailureCode(response.text()):TaskFinalizationGate.ParseFailureCode.NONE; @@ -205,16 +231,16 @@ public final class TaskFinalizationCoordinator { outcome=evaluated.outcome().name(); diagnostic=diagnostic(verdict,evaluated,ledger,attempts); if (evaluated.outcome() == TaskFinalizationGate.Outcome.INCOMPLETE) - return incomplete(diagnostic.name(), response, 1, false); - return new Review(evaluated,response,1,false); - } catch(CancellationException cancelled) {diagnostic=Diagnostic.CANCELLED;return incomplete("Cancelled",response,1,true);} - catch(Exception unavailable) {diagnostic=Diagnostic.CHECKER_EXCEPTION;return incomplete("CHECKER_EXCEPTION",response,1,false);} + return incomplete(diagnostic.name(), usage, calls, false); + return new Review(evaluated,usage,calls,false); + } catch(CancellationException cancelled) {diagnostic=Diagnostic.CANCELLED;return incomplete("Cancelled",usage,calls,true);} + catch(Exception unavailable) {diagnostic=Diagnostic.CHECKER_EXCEPTION;return incomplete("CHECKER_EXCEPTION",usage,calls,false);} finally { - var completed=new java.util.LinkedHashMap(Map.of("logicalCalls",1,"promptTokens",response==null?0:response.promptTokens(), - "completionTokens",response==null?0:response.completionTokens(),"usageReported",response!=null, - "cacheReadTokens",response==null?0:response.cacheReadTokens(), - "cacheWriteTokens",response==null?0:response.cacheWriteTokens(), - "reasoningTokens",response==null?0:response.reasoningTokens(),"outcome",outcome)); + var completed=new java.util.LinkedHashMap(Map.of("logicalCalls",calls,"promptTokens",usage==null?0:usage.promptTokens(), + "completionTokens",usage==null?0:usage.completionTokens(),"usageReported",usage!=null, + "cacheReadTokens",usage==null?0:usage.cacheReadTokens(), + "cacheWriteTokens",usage==null?0:usage.cacheWriteTokens(), + "reasoningTokens",usage==null?0:usage.reasoningTokens(),"outcome",outcome)); completed.put("diagnostic",diagnostic.name()); completed.put("parseFailureCode",parseFailureCode.name()); completed.put("responseTextChars",response==null || response.text()==null?0:response.text().length()); @@ -256,19 +282,22 @@ public final class TaskFinalizationCoordinator { return options; } public static ChatOptions toolFreeOptions(ChatModel model) { + return toolFreeOptions(model,OUTPUT_TOKENS); + } + private static ChatOptions toolFreeOptions(ChatModel model,int outputTokens) { ChatOptions defaults=ChatModelDefaults.optionsFor(model); - ChatOptions options=defaults==null?ChatOptions.builder().maxTokens(OUTPUT_TOKENS).build():defaults.mutate().maxTokens(OUTPUT_TOKENS).build(); + ChatOptions options=defaults==null?ChatOptions.builder().maxTokens(outputTokens).build():defaults.mutate().maxTokens(outputTokens).build(); if(options instanceof ToolCallingChatOptions tools) options=tools.mutate().toolCallbacks(List.of()).build(); // OpenAI reasoning providers use maxCompletionTokens instead of maxTokens. if(options instanceof org.springframework.ai.openai.OpenAiChatOptions openAi) { options = openAi.getMaxCompletionTokens()!=null - ? openAi.mutate().maxTokens(null).maxCompletionTokens(OUTPUT_TOKENS).toolChoice("none").build() - : openAi.mutate().maxTokens(OUTPUT_TOKENS).toolChoice("none").build(); + ? openAi.mutate().maxTokens(null).maxCompletionTokens(outputTokens).toolChoice("none").build() + : openAi.mutate().maxTokens(outputTokens).toolChoice("none").build(); } if(options instanceof org.springframework.ai.anthropic.AnthropicChatOptions anthropic) - options=anthropic.mutate().thinkingDisabled().maxTokens(OUTPUT_TOKENS).toolCallbacks(List.of()).build(); + options=anthropic.mutate().thinkingDisabled().maxTokens(outputTokens).toolCallbacks(List.of()).build(); if(options instanceof vip.mate.llm.dashscope.NativeDashScopeChatOptions dashscope) - options=dashscope.mutate().enableThinking(false).maxTokens(OUTPUT_TOKENS).toolCallbacks(List.of()).build(); + options=dashscope.mutate().enableThinking(false).maxTokens(outputTokens).toolCallbacks(List.of()).build(); return options; } private void event(String id,String status,Map details) { @@ -295,6 +324,7 @@ public final class TaskFinalizationCoordinator { case "Invalid verification facts", "INVALID_VERDICT", "FAILED_CALL_COUNT_MISMATCH", "UNSUPPORTED_RECOVERY", "CONTRADICTORY_OUTCOME", "INVALID_CONTINUATION", "VERIFICATION_REJECTED" -> "最终核验返回格式无效或与实际工具记录不一致，未发布未经核实的回答"; case "CHECKER_PARTIAL" -> "最终核验响应被截断，无法确认核验结果"; + case "CHECKER_OUTPUT_LIMIT" -> "最终核验在生成核验结果前耗尽输出 token 预算，请适当增加模型最大输出 token 数或拆分任务后重试"; case "CHECKER_TOOL_CALL" -> "最终核验意外返回工具调用，未获得有效核验结果"; case "Checker unavailable", "CHECKER_ERROR", "CHECKER_EXCEPTION", "Verification unavailable", "Summary verification unavailable", "Summary unavailable" -> "最终核验或汇总调用失败，任务是否完成尚未确认"; diff --git a/mateclaw-server/src/test/java/vip/mate/agent/graph/NodeStreamingChatHelperThinkingCapTest.java b/mateclaw-server/src/test/java/vip/mate/agent/graph/NodeStreamingChatHelperThinkingCapTest.java index 67da4bcd..52a04304 100644 --- a/mateclaw-server/src/test/java/vip/mate/agent/graph/NodeStreamingChatHelperThinkingCapTest.java +++ b/mateclaw-server/src/test/java/vip/mate/agent/graph/NodeStreamingChatHelperThinkingCapTest.java @@ -47,6 +47,21 @@ class NodeStreamingChatHelperThinkingCapTest { return m; } + @Test + void lengthFinishIsRetainedEvenWhenPartialContentWasEmitted() { + var msg = AssistantMessage.builder().content("{\"decision\":\"ACCEPT\",") + .properties(Map.of("reasoningContent", "Reviewing fifteen forecast rows")).build(); + var metadata = ChatGenerationMetadata.builder().finishReason("length").build(); + var model = mock(ChatModel.class); + when(model.stream(any(Prompt.class))).thenReturn(Flux.just( + new ChatResponse(List.of(new Generation(msg, metadata))))); + var result = new NodeStreamingChatHelper(streamTracker) + .streamCallSilent(model, smallPrompt(), "c", "finalization_check"); + assertEquals("length", result.assistantMessage().getMetadata().get("generationFinishReason")); + assertEquals(msg.getText(), result.text()); + assertFalse(result.partial(), "Do not change live chunk/continuation semantics when retaining finish metadata"); + } + @Test @DisplayName("thinking >= cap + tool_call in same chunk: cap must NOT trigger; tool_call survives") void thinkingAndToolCallSameChunk_doesNotTripSoftCap() { diff --git a/mateclaw-server/src/test/java/vip/mate/agent/graph/guard/TaskFinalizationCoordinatorTest.java b/mateclaw-server/src/test/java/vip/mate/agent/graph/guard/TaskFinalizationCoordinatorTest.java index 8a12e283..2afd43b7 100644 --- a/mateclaw-server/src/test/java/vip/mate/agent/graph/guard/TaskFinalizationCoordinatorTest.java +++ b/mateclaw-server/src/test/java/vip/mate/agent/graph/guard/TaskFinalizationCoordinatorTest.java @@ -26,6 +26,135 @@ class TaskFinalizationCoordinatorTest { static String json(String decision,String claims){return "{\"decision\":\""+decision+"\",\"taskComplete\":true,\"remainingWork\":false,\"failedCallCount\":1,\"recoveryClaims\":"+claims+",\"reason\":\"checked\"}";} NodeStreamingChatHelper.StreamResult result(String text){return new NodeStreamingChatHelper.StreamResult(text,"checker thinking",null,List.of(),false,7,3,false,null,NodeStreamingChatHelper.ErrorType.NONE,false,2,1,1);} TaskFinalizationCoordinator.Review review(int window,int attempts,int used){return coordinator.review(model,"c","Calculate all stages","candidate",ledger,"untrusted progress",window,attempts,10,used,true,new RunState());} + @Test void reasoningTokenExhaustionRetriesOnceWithConfiguredBudget() { + var defaults = OpenAiChatOptions.builder().model("sensenova-6.8-flash-lite").maxTokens(8192).build(); + when(model.getDefaultOptions()).thenReturn(defaults); + var calls = new java.util.concurrent.atomic.AtomicInteger(); + when(helper.streamCallSilent(any(), any(), anyString(), anyString())).thenAnswer(call -> { + Prompt prompt = call.getArgument(1); + assertEquals("off", ThinkingLevelHolder.get()); + assertTrue(((ToolCallingChatOptions) prompt.getOptions()).getToolCallbacks().isEmpty()); + if (calls.getAndIncrement() == 0) { + assertEquals(1024, prompt.getOptions().getMaxTokens()); + return new NodeStreamingChatHelper.StreamResult("", "Reviewing all fifteen forecast days", null, + List.of(), false, 10, 1024, true, "thinking_token_limit", + NodeStreamingChatHelper.ErrorType.NONE, false, 2, 0, 1024); + } + assertEquals(8192, prompt.getOptions().getMaxTokens()); + return result(json("ACCEPT", "[]")); + }); + var check = review(16000, 0, 0); + assertEquals(TaskFinalizationGate.Outcome.ACCEPT, check.result().outcome()); + assertEquals(2, check.calls()); + assertEquals(17, check.usage().promptTokens()); + assertEquals(1027, check.usage().completionTokens()); + assertEquals(4, check.usage().cacheReadTokens()); + assertEquals(1025, check.usage().reasoningTokens()); + assertEquals(8192, defaults.getMaxTokens()); + } + + @Test void tokenRetryStaysWithinWindowAndKeepsObjectiveAndEvidence() { + when(model.getDefaultOptions()).thenReturn(OpenAiChatOptions.builder().maxTokens(32000).build()); + var calls = new java.util.concurrent.atomic.AtomicInteger(); + when(helper.streamCallSilent(any(), any(), anyString(), anyString())).thenAnswer(call -> { + Prompt prompt = call.getArgument(1); + assertTrue(prompt.getContents().contains("Calculate all stages")); + assertTrue(prompt.getContents().contains("missing language")); + assertTrue(vip.mate.agent.context.TokenEstimator.estimateTokens(prompt.getInstructions()) + + prompt.getOptions().getMaxTokens() <= 8000); + if (calls.getAndIncrement() == 0) { + return new NodeStreamingChatHelper.StreamResult("", "Review in progress", null, + List.of(), false, 10, 1024, true, "thinking_token_limit", + NodeStreamingChatHelper.ErrorType.NONE, false, 0, 0, 1024); + } + assertTrue(prompt.getOptions().getMaxTokens() > 1024); + return result(json("ACCEPT", "[]")); + }); + assertEquals(TaskFinalizationGate.Outcome.ACCEPT, review(8000, 0, 0).result().outcome()); + assertEquals(2, calls.get()); + } + + @Test void exhaustedCallBudgetReportsTokenLimitWithoutRetrying() { + when(model.getDefaultOptions()).thenReturn(OpenAiChatOptions.builder().maxTokens(8192).build()); + when(helper.streamCallSilent(any(), any(), anyString(), anyString())).thenReturn( + new NodeStreamingChatHelper.StreamResult("", "Review in progress", null, + List.of(), false, 10, 1024, true, "thinking_token_limit", + NodeStreamingChatHelper.ErrorType.NONE, false, 0, 0, 1024)); + var check = review(16000, 0, 49); + assertEquals(TaskFinalizationGate.Outcome.INCOMPLETE, check.result().outcome()); + assertEquals("CHECKER_OUTPUT_LIMIT", check.result().reason()); + assertEquals(1, check.calls()); + assertTrue(TaskFinalizationCoordinator.incompleteAnswer(check.result().reason()).contains("token")); + verify(helper, times(1)).streamCallSilent(any(), any(), anyString(), anyString()); + } + + @Test void repeatedExhaustionIsBoundedAndReportsAggregateUsage() { + when(model.getDefaultOptions()).thenReturn(OpenAiChatOptions.builder().maxTokens(32000).build()); + var tracker = mock(vip.mate.channel.web.ChatStreamTracker.class); + var observed = new TaskFinalizationCoordinator(helper, tracker); + when(helper.streamCallSilent(any(), any(), anyString(), anyString())).thenReturn( + new NodeStreamingChatHelper.StreamResult("", "Still reviewing", null, + List.of(), false, 10, 1024, true, "thinking_token_limit", + NodeStreamingChatHelper.ErrorType.NONE, false, 2, 1, 1024)); + var check = observed.review(model, "c", "objective", "candidate", ledger, "", 16000, 0, 10, 0, true, new RunState()); + assertEquals(TaskFinalizationGate.Outcome.INCOMPLETE, check.result().outcome()); + assertEquals("CHECKER_OUTPUT_LIMIT", check.result().reason()); + assertEquals(2, check.calls()); + assertEquals(20, check.usage().promptTokens()); + assertEquals(2048, check.usage().completionTokens()); + var prompts = org.mockito.ArgumentCaptor.forClass(Prompt.class); + verify(helper, times(2)).streamCallSilent(any(), prompts.capture(), anyString(), anyString()); + assertEquals(8192, prompts.getAllValues().get(1).getOptions().getMaxTokens()); + var events = org.mockito.ArgumentCaptor.forClass(Map.class); + verify(tracker, times(2)).broadcastObject(eq("c"), eq("finalization_check"), events.capture()); + var completed = events.getAllValues().get(1); + assertEquals(2, completed.get("logicalCalls")); + assertEquals(20, completed.get("promptTokens")); + assertEquals(2048, completed.get("completionTokens")); + assertEquals("CHECKER_OUTPUT_LIMIT", completed.get("diagnostic")); + } + + @Test void retryCannotTurnInvalidVerdictIntoAcceptance() { + when(model.getDefaultOptions()).thenReturn(OpenAiChatOptions.builder().maxTokens(8192).build()); + when(helper.streamCallSilent(any(), any(), anyString(), anyString())).thenReturn( + new NodeStreamingChatHelper.StreamResult("", "Review", null, List.of(), false, + 10, 1024, true, "thinking_token_limit", NodeStreamingChatHelper.ErrorType.NONE, + false, 0, 0, 1024), result("not a verdict")); + var check = review(16000, 0, 0); + assertEquals(TaskFinalizationGate.Outcome.INCOMPLETE, check.result().outcome()); + assertEquals("INVALID_VERDICT", check.result().reason()); + assertEquals(2, check.calls()); + } + + @Test void noBudgetGrowthAndSafetyCapsAreNotRetried() { + for (String reason : List.of("thinking_token_limit", "thinking_only_no_content", "content_repetition")) { + reset(helper); + when(model.getDefaultOptions()).thenReturn(OpenAiChatOptions.builder().maxTokens(1024).build()); + when(helper.streamCallSilent(any(), any(), anyString(), anyString())).thenReturn( + new NodeStreamingChatHelper.StreamResult("", "Review", null, List.of(), false, + 10, 1024, true, reason, NodeStreamingChatHelper.ErrorType.NONE, false, 0, 0, 1024)); + assertEquals(TaskFinalizationGate.Outcome.INCOMPLETE, review(16000, 0, 0).result().outcome()); + verify(helper, times(1)).streamCallSilent(any(), any(), anyString(), anyString()); + } + } + + @Test void completionTokenRetryPreservesOpenAiParameterFamily() { + when(model.getDefaultOptions()).thenReturn(OpenAiChatOptions.builder().model("gpt-5") + .maxCompletionTokens(8192).build()); + var calls = new java.util.concurrent.atomic.AtomicInteger(); + when(helper.streamCallSilent(any(), any(), anyString(), anyString())).thenAnswer(call -> { + var options = (OpenAiChatOptions) ((Prompt) call.getArgument(1)).getOptions(); + assertNull(options.getMaxTokens()); + assertEquals(calls.get() == 0 ? 1024 : 8192, options.getMaxCompletionTokens()); + if (calls.getAndIncrement() == 0) return new NodeStreamingChatHelper.StreamResult("", "Review", null, + List.of(), false, 10, 1024, true, "thinking_token_limit", NodeStreamingChatHelper.ErrorType.NONE, + false, 0, 0, 1024); + return result(json("ACCEPT", "[]")); + }); + assertEquals(TaskFinalizationGate.Outcome.ACCEPT, review(16000, 0, 0).result().outcome()); + assertEquals(2, calls.get()); + } + @Test void incompleteExplanationDistinguishesFixedCausesWithoutLeakingReviewerText() { var context=TaskFinalizationCoordinator.incompleteAnswer("Required execution facts or objective/candidate cannot fit context"); var correction=TaskFinalizationCoordinator.incompleteAnswer("Correction allowance exhausted"); diff --git a/mateclaw-server/src/test/java/vip/mate/llm/chatmodel/ReasoningFinalizationTransportTest.java b/mateclaw-server/src/test/java/vip/mate/llm/chatmodel/ReasoningFinalizationTransportTest.java new file mode 100644 index 00000000..f1b0dcda --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/llm/chatmodel/ReasoningFinalizationTransportTest.java @@ -0,0 +1,113 @@ +package vip.mate.llm.chatmodel; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.sun.net.httpserver.HttpServer; +import io.micrometer.observation.ObservationRegistry; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.support.StaticListableBeanFactory; +import org.springframework.retry.support.RetryTemplate; +import org.springframework.web.client.RestClient; +import org.springframework.web.reactive.function.client.WebClient; +import vip.mate.agent.GraphEventPublisher; +import vip.mate.agent.graph.NodeStreamingChatHelper; +import vip.mate.agent.graph.guard.TaskFinalizationCoordinator; +import vip.mate.agent.graph.guard.TaskFinalizationGate; +import vip.mate.agent.graph.state.ActionExecutionLedger; +import vip.mate.agent.runtime.state.RunState; +import vip.mate.channel.web.ChatStreamTracker; +import vip.mate.llm.model.ModelConfigEntity; +import vip.mate.llm.model.ModelProviderEntity; +import vip.mate.llm.service.ModelProviderService; + +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; + +/** Exercises the real SDK and SSE helper against a reasoning-provider protocol fixture. */ +class ReasoningFinalizationTransportTest { + private static final ObjectMapper JSON = new ObjectMapper(); + + @Test + void reasoningOnlyLengthResponseRecoversBeforePublishingVerdict() throws Exception { + assertRecovers(""); + } + + @Test + void lengthResponseWithTruncatedJsonRecoversBeforePublishingVerdict() throws Exception { + assertRecovers("{\"decision\":\"ACCEPT\",\"taskComplete\":true,"); + } + + private void assertRecovers(String exhaustedContent) throws Exception { + var requests = new CopyOnWriteArrayList(); + var server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/v1/chat/completions", exchange -> { + JsonNode request = JSON.readTree(exchange.getRequestBody()); + requests.add(request); + boolean exhausted = request.path("max_tokens").asInt() <= 1024; + String verdict = "{\"decision\":\"ACCEPT\",\"taskComplete\":true,\"remainingWork\":false," + + "\"failedCallCount\":0,\"recoveryClaims\":[],\"reason\":\"Forecast matches tool evidence\"}"; + String content = exhausted ? exhaustedContent : verdict; + // SenseNova documents reasoning (rather than reasoning_content) and + // an empty finish_reason on intermediate chunks. max_tokens includes reasoning. + String events = "data: " + chunk(Map.of("reasoning", "Checking the fifteen daily forecasts"), "") + + "\n\ndata: " + chunk(Map.of("content", content), exhausted ? "length" : "stop") + + "\n\ndata: {\"id\":\"fixture\",\"object\":\"chat.completion.chunk\",\"created\":1," + + "\"model\":\"sensenova-6.8-flash-lite\",\"choices\":[],\"usage\":{\"prompt_tokens\":200," + + "\"completion_tokens\":" + (exhausted ? 1024 : 1400) + + ",\"total_tokens\":" + (exhausted ? 1224 : 1600) + "}}\n\ndata: [DONE]\n\n"; + byte[] body = events.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type", "text/event-stream; charset=utf-8"); + exchange.sendResponseHeaders(200, body.length); + try (var out = exchange.getResponseBody()) { out.write(body); } + }); + server.start(); + try { + var service = mock(ModelProviderService.class); + var provider = new ModelProviderEntity(); + provider.setProviderId("sensenova-fixture"); + provider.setRequireApiKey(false); + provider.setBaseUrl("http://127.0.0.1:" + server.getAddress().getPort()); + when(service.isProviderConfigured(provider.getProviderId())).thenReturn(true); + when(service.readProviderGenerateKwargs(provider)).thenReturn(Map.of()); + var beans = new StaticListableBeanFactory(); + beans.addBean("observations", ObservationRegistry.NOOP); + var builder = new OpenAiCompatibleChatModelBuilder(service, + beans.getBeanProvider(RestClient.Builder.class), beans.getBeanProvider(WebClient.Builder.class), + beans.getBeanProvider(ObservationRegistry.class)); + var config = new ModelConfigEntity(); + config.setModelName("sensenova-6.8-flash-lite"); + config.setMaxTokens(8192); + var model = builder.build(config, provider, RetryTemplate.builder().maxAttempts(1).build()); + var tracker = mock(ChatStreamTracker.class); + var helper = new NodeStreamingChatHelper(tracker); + var coordinator = new TaskFinalizationCoordinator(helper, tracker); + var ledger = ActionExecutionLedger.fromEvents(List.of( + GraphEventPublisher.toolComplete("forecast", "execute_code", "15 days of Shanghai forecast data", true))); + var result = coordinator.review(model, "test-672", "上海未来15天天气", "Forecast summary", ledger, + "", 16000, 0, 8, 3, true, new RunState()); + assertEquals(TaskFinalizationGate.Outcome.ACCEPT, result.result().outcome()); + assertEquals(2, result.calls()); + assertEquals(2, requests.size()); + assertEquals(1024, requests.get(0).path("max_tokens").asInt()); + assertEquals(8192, requests.get(1).path("max_tokens").asInt()); + assertEquals(requests.get(0).path("messages"), requests.get(1).path("messages")); + assertTrue(requests.stream().allMatch(request -> request.path("tools").isEmpty())); + assertNull(ThinkingLevelHolder.get()); + } finally { + ThinkingLevelHolder.clear(); + server.stop(0); + } + } + + private static String chunk(Map delta, String finish) throws java.io.IOException { + return JSON.writeValueAsString(Map.of("id", "fixture", "object", "chat.completion.chunk", "created", 1, + "model", "sensenova-6.8-flash-lite", "choices", List.of(Map.of("index", 0, "delta", delta, + "finish_reason", finish)))); + } +}