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 fc606e56..1b2c0365 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 @@ -46,13 +46,18 @@ public final class TaskFinalizationCoordinator { Endorse recovery only for canonical supportedRecoveryCallIds; an empty list means none is established. Do not invent numeric results or reconstruct omitted results. Unsupported candidate prose requires REVISE with empty recoveryClaims and a specific correction. - Unfinished executable objective work requires CONTINUE; do not automatically replay failed calls. + Unfinished executable objective work or unsatisfied requested state constraints require CONTINUE, not prose-only REVISE. + Explicitly requested pending records are not unfinished executable work; compare them to the original constraints. + If a record requested unchanged was modified, request its authorized restoration with CONTINUE; do not merely revise the narrative. + Do not automatically replay failed calls. An empty recoveryClaims does not excuse unsupported candidate prose. Inspect every prose, table and JSON claim independently. If the candidate implies an unsupported retry, parameter-fix success or original-call recovery, return REVISE with empty recoveryClaims, even if its task-level recovery flag is valid. Return ONLY one JSON object with exactly these six fields. No markdown fences, preamble, trailing text, or extra explanations. Decision choices: ACCEPT, REVISE, CONTINUE. Choose exactly one value, never a combined value. The reason must be at most 256 characters: one short evidence-based sentence, without per-stage calculations or receipt lists. + For REVISE, quote the short offending clause and give a concrete grounded replacement in that same reason. + Distinguish a false factual claim from truthful negation or explicitly scoped later successes; do not reject those merely for containing recovery words. Valid ACCEPT example (replace values with the actual verified facts): {"decision":"ACCEPT","taskComplete":true,"remainingWork":false, "failedCallCount":0,"recoveryClaims":[],"reason":"All requested stages and aggregate match canonical outputs."} diff --git a/mateclaw-server/src/main/java/vip/mate/agent/graph/guard/TaskFinalizationGate.java b/mateclaw-server/src/main/java/vip/mate/agent/graph/guard/TaskFinalizationGate.java index 43b6628b..477c2c0a 100644 --- a/mateclaw-server/src/main/java/vip/mate/agent/graph/guard/TaskFinalizationGate.java +++ b/mateclaw-server/src/main/java/vip/mate/agent/graph/guard/TaskFinalizationGate.java @@ -151,7 +151,11 @@ public final class TaskFinalizationGate { } if (verdict.decision() == Outcome.ACCEPT) return new Result(Outcome.ACCEPT, verdict.reason()); if (attempts >= MAX_CORRECTIVE_ATTEMPTS || !canContinue) return incomplete("Correction allowance exhausted"); - return new Result(verdict.decision(), verdict.reason()); + // Work-state facts take precedence over a model's prose-only repair label. + // This still consumes the same bounded correction allowance and never accepts a draft. + Outcome next = verdict.decision() == Outcome.REVISE && !verdict.taskComplete() && verdict.remainingWork() + ? Outcome.CONTINUE : verdict.decision(); + return new Result(next, verdict.reason()); } private static Result incomplete(String reason) { return new Result(Outcome.INCOMPLETE, reason); } 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 3cdabf83..ffe22ceb 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,16 @@ 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 revisionFeedbackRequestsSpecificOffendingClauseAndGroundedReplacement() { + when(helper.streamCallSilent(any(),any(),anyString(),anyString())).thenAnswer(call->{ + String prompt=((Prompt)call.getArgument(1)).getContents(); + assertTrue(prompt.contains("quote the short offending clause")); + assertTrue(prompt.contains("give a concrete grounded replacement")); + assertTrue(prompt.contains("at most 256 characters")); + return result(json("REVISE","[]")); + }); + assertEquals(TaskFinalizationGate.Outcome.REVISE,review(16000,0,0).result().outcome()); + } @Test void latestAggregateAfterPreviewIsGroundedFromExactRetainedResponse() { String output="x".repeat(600)+" totalSquareSum=2094116873200"; var receipts=ActionExecutionLedger.fromEvents(List.of(GraphEventPublisher.toolComplete("aggregate","execute_code",output,true))); diff --git a/mateclaw-server/src/test/java/vip/mate/agent/graph/guard/TaskFinalizationGateTest.java b/mateclaw-server/src/test/java/vip/mate/agent/graph/guard/TaskFinalizationGateTest.java index 8302a184..c0d74ee4 100644 --- a/mateclaw-server/src/test/java/vip/mate/agent/graph/guard/TaskFinalizationGateTest.java +++ b/mateclaw-server/src/test/java/vip/mate/agent/graph/guard/TaskFinalizationGateTest.java @@ -113,10 +113,21 @@ class TaskFinalizationGateTest { assertEquals(TaskFinalizationGate.Outcome.REVISE,gate.evaluate(revision,ActionExecutionLedger.empty(),0,false,true).outcome()); } + @Test void unfinishedRevisionRoutesToExecutionRatherThanProseOnlyRepair() { + var verdict=gate.parse(json("REVISE",false,true,0,"[]")).orElseThrow(); + assertEquals(TaskFinalizationGate.Outcome.CONTINUE, + gate.evaluate(verdict,ActionExecutionLedger.empty(),0,false,true).outcome()); + for(int attempts:List.of(0,1)) { + assertEquals(TaskFinalizationGate.Outcome.INCOMPLETE, + gate.evaluate(verdict,ActionExecutionLedger.empty(),attempts,false,false).outcome()); + } + assertEquals(TaskFinalizationGate.Outcome.INCOMPLETE, + gate.evaluate(verdict,ActionExecutionLedger.empty(),2,false,true).outcome()); + } @Test void attemptsAreTotalAndTerminalStateTakesPrecedence() { for(String decision:List.of("CONTINUE","REVISE")) { var verdict=gate.parse(json(decision,false,true,0,"[]")).orElseThrow(); - assertEquals(TaskFinalizationGate.Outcome.valueOf(decision),gate.evaluate(verdict,ActionExecutionLedger.empty(),1,false,true).outcome()); + assertEquals(TaskFinalizationGate.Outcome.CONTINUE,gate.evaluate(verdict,ActionExecutionLedger.empty(),1,false,true).outcome()); assertEquals(TaskFinalizationGate.Outcome.INCOMPLETE,gate.evaluate(verdict,ActionExecutionLedger.empty(),2,false,true).outcome()); assertEquals(TaskFinalizationGate.Outcome.INCOMPLETE,gate.evaluate(verdict,ActionExecutionLedger.empty(),0,true,true).outcome()); assertEquals(TaskFinalizationGate.Outcome.INCOMPLETE,gate.evaluate(verdict,ActionExecutionLedger.empty(),0,false,false).outcome()); diff --git a/mateclaw-server/src/test/java/vip/mate/agent/graph/node/PayloadWireIntegrationTest.java b/mateclaw-server/src/test/java/vip/mate/agent/graph/node/PayloadWireIntegrationTest.java new file mode 100644 index 00000000..891834b1 --- /dev/null +++ b/mateclaw-server/src/test/java/vip/mate/agent/graph/node/PayloadWireIntegrationTest.java @@ -0,0 +1,137 @@ +package vip.mate.agent.graph.node; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.sun.net.httpserver.HttpServer; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.springframework.ai.chat.messages.*; +import vip.mate.agent.AgentToolSet; +import vip.mate.agent.context.PrefixBudgetPlan; +import vip.mate.agent.graph.NodeStreamingChatHelper; +import vip.mate.agent.graph.PayloadReduction; +import vip.mate.agent.runtime.state.RunState; +import vip.mate.channel.web.ChatStreamTracker; +import vip.mate.llm.chatmodel.OpenAiCompatibleChatModelBuilder; +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.*; +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; +import static vip.mate.agent.graph.state.MateClawStateKeys.*; + +/** Actual HTTP bodies/statuses through the production provider and reasoning recovery. */ +class PayloadWireIntegrationTest { + @ParameterizedTest + @ValueSource(strings={"recover", "mandatory", "stillTooLarge", "reviewTooLarge", "unfinishedRevision"}) + void boundsWireRetriesAndPreservesMandatoryUtf8Input(String mode) throws Exception { + var bodies = Collections.synchronizedList(new ArrayList()); + var server = HttpServer.create(new InetSocketAddress("127.0.0.1",0),0); + server.createContext("/", exchange -> { + byte[] request = exchange.getRequestBody().readAllBytes(); + bodies.add(request); + boolean reject = mode.equals("reviewTooLarge") ? bodies.size()>1 + : !mode.equals("unfinishedRevision") && request.length > (mode.equals("stillTooLarge") ? 64 : 40_000); + String response = reject + ? "{\"error\":{\"message\":\"body limit exceeded\",\"type\":\"invalid_request_error\"}}" + : "data: {\"id\":\"r\",\"object\":\"chat.completion.chunk\",\"created\":1,\"model\":\"fixture\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"verified result\"},\"finish_reason\":\"stop\"}]}\n\ndata: [DONE]\n\n"; + if(mode.equals("unfinishedRevision") && bodies.size()>1) { + String verdict="{\"decision\":\"REVISE\",\"taskComplete\":false,\"remainingWork\":true,\"failedCallCount\":1,\"recoveryClaims\":[],\"reason\":\"Restore the requested pending checkpoint before answering\"}"; + response=response.replace("\"verified result\"",new ObjectMapper().writeValueAsString(verdict)); + } + byte[] bytes=response.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type",reject?"application/json":"text/event-stream"); + exchange.sendResponseHeaders(reject?413:200,bytes.length); + exchange.getResponseBody().write(bytes); + exchange.close(); + }); + server.start(); + try { + var service=mock(ModelProviderService.class); + when(service.isProviderConfigured(anyString())).thenReturn(true); + when(service.readProviderGenerateKwargs(any())).thenReturn(Map.of()); + var config=new ModelConfigEntity(); config.setModelName("fixture"); config.setRequestTimeoutSeconds(3); + var provider=new ModelProviderEntity(); provider.setProviderId("fixture"); provider.setRequireApiKey(false); + provider.setBaseUrl("http://127.0.0.1:"+server.getAddress().getPort()); + var model=new OpenAiCompatibleChatModelBuilder(service,null,null,null).build(config,provider,null); + String schema="{\"type\":\"object\",\"properties\":{\"language\":{\"type\":\"string\"},\"code\":{\"type\":\"string\"},\"args\":{\"type\":\"string\"}},\"required\":[\"language\",\"code\"],\"additionalProperties\":false}"; + var callback=new org.springframework.ai.tool.ToolCallback() { + public org.springframework.ai.tool.definition.ToolDefinition getToolDefinition() { + return org.springframework.ai.tool.definition.ToolDefinition.builder().name("execute_code") + .description("Synthetic fixture only").inputSchema(schema).build(); + } + public String call(String arguments) { throw new AssertionError("History recovery must not replay tools"); } + }; + var node=new ReasoningNode(model,AgentToolSet.fromCallbacks(List.of(),List.of(callback)),null,false, + new NodeStreamingChatHelper(mock(ChatStreamTracker.class)),null,null,1024,null,null,null,null); + node.setPrefixBudgetPlan(new PrefixBudgetPlan(true,1_000_000,PrefixBudgetPlan.Profile.NORMAL,0,0,0,0,0,0,0)); + String objective="必要输入：保持原文与引号 \"参数\" 🧪"; + if (mode.equals("mandatory")) objective = objective.repeat(2000); + String originalArguments=new ObjectMapper().writeValueAsString(Map.of( + "language","python","code","print(\"历史🧪\")","args","[\"空 格\",\"back\\\\slash\"]")); + List canonical=new ArrayList<>(); canonical.add(new UserMessage(objective)); + if (!mode.equals("mandatory")) for(int i=0;i<20;i++) { + String id="wire-"+i; + canonical.add(AssistantMessage.builder().content("").toolCalls(List.of( + new AssistantMessage.ToolCall(id,"function","execute_code",originalArguments))).build()); + canonical.add(ToolResponseMessage.builder().responses(List.of( + new ToolResponseMessage.ToolResponse(id,"execute_code","历史🧪".repeat(500)))).build()); + } + int originalSize=canonical.size(); + String originalHistory=canonical.toString(); + var values=new HashMap(); + boolean guardedReview=mode.equals("reviewTooLarge") || mode.equals("unfinishedRevision"); + if(guardedReview) { + values.put(CURRENT_ITERATION,2); + var events=List.of(vip.mate.agent.GraphEventPublisher.toolComplete("f","calculate","missing language",false), + vip.mate.agent.GraphEventPublisher.toolComplete("s","calculate","42",true)); + values.put(PENDING_EVENTS,events); + values.put(ACTION_EXECUTION_LEDGER,vip.mate.agent.graph.state.ActionExecutionLedger.fromEvents(events)); + node.setFinalizationCoordinator(new vip.mate.agent.graph.guard.TaskFinalizationCoordinator( + new NodeStreamingChatHelper(mock(ChatStreamTracker.class)),null)); + } + values.putAll(Map.of(CONVERSATION_ID,"wire-"+mode,SYSTEM_PROMPT,"MANDATORY_SKILL", + USER_MESSAGE,objective,MESSAGES,canonical,MAX_ITERATIONS,100)); + var output=node.apply(new RunState(values)); + int expected=mode.equals("mandatory")?1:2; + assertEquals(expected,bodies.size(),"no SDK/helper/fallback or unchanged payload retries"); + assertEquals(expected,output.get(LLM_CALL_COUNT)); + assertEquals(originalSize,canonical.size(),"canonical history must survive local reduction"); + assertEquals(originalHistory,canonical.toString(),"history contents and arguments must remain intact"); + if(expected==2 && !guardedReview) assertTrue(bodies.get(1).length calls=new HashSet<>(),responses=new HashSet<>(); + for(JsonNode message:messages) { + hasObjective |= message.path("content").asText().contains(objective); + hasSkill |= message.path("content").asText().contains("MANDATORY_SKILL"); + for(JsonNode call:message.path("tool_calls")) { + calls.add(call.path("id").asText()); + assertEquals(originalArguments,call.at("/function/arguments").asText()); + } + if(message.path("role").asText().equals("tool")) responses.add(message.path("tool_call_id").asText()); + } + assertTrue(hasObjective); + if(!guardedReview || body==bodies.get(0)) assertTrue(hasSkill); + assertEquals(calls,responses); + if(!guardedReview || body==bodies.get(0)) + assertEquals(json.readTree(schema),json.readTree(body).at("/tools/0/function/parameters")); + } + if(guardedReview) { + if(mode.equals("unfinishedRevision")) { + assertEquals(true,output.get(CONTINUE_REASONING)); + assertTrue(String.valueOf(output.get(MESSAGES)).contains("Requested outcome: CONTINUE")); + assertEquals(1,output.get(FINALIZATION_ATTEMPTS)); + } else assertEquals("incomplete",output.get(FINISH_REASON)); + assertFalse(String.valueOf(output.get(FINAL_ANSWER)).contains("verified result")); + assertEquals(false,output.get(CONTENT_STREAMED)); + assertFalse(json.readTree(bodies.get(1)).has("tools"),"review must not replay tools"); + } else assertEquals(mode.equals("recover")?"verified result":"[错误] "+PayloadReduction.INCOMPLETE,output.get(FINAL_ANSWER)); + } finally { server.stop(0); } + } +} diff --git a/mateclaw-server/src/test/java/vip/mate/config/SchedulingConfigPausedShutdownTest.java b/mateclaw-server/src/test/java/vip/mate/config/SchedulingConfigPausedShutdownTest.java index 152edce6..44bc17e5 100644 --- a/mateclaw-server/src/test/java/vip/mate/config/SchedulingConfigPausedShutdownTest.java +++ b/mateclaw-server/src/test/java/vip/mate/config/SchedulingConfigPausedShutdownTest.java @@ -95,12 +95,11 @@ class SchedulingConfigPausedShutdownTest { private static void awaitActiveWorkers(ThreadPoolTaskScheduler scheduler, int expected) throws Exception { long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2); - while (scheduler.getScheduledThreadPoolExecutor().getActiveCount() < expected - && System.nanoTime() < deadline) { + while (System.nanoTime() < deadline) { + // Keep the successful observation: a second sample can race a worker transition. + if (scheduler.getScheduledThreadPoolExecutor().getActiveCount() >= expected) return; Thread.sleep(5); } - assertTrue( - scheduler.getScheduledThreadPoolExecutor().getActiveCount() >= expected, - "pending tick must reach the lifecycle pause gate"); + fail("pending tick must reach the lifecycle pause gate"); } } diff --git a/mateclaw-server/src/test/java/vip/mate/config/ServletJsonContractTest.java b/mateclaw-server/src/test/java/vip/mate/config/ServletJsonContractTest.java index 30e13633..a997ed0e 100644 --- a/mateclaw-server/src/test/java/vip/mate/config/ServletJsonContractTest.java +++ b/mateclaw-server/src/test/java/vip/mate/config/ServletJsonContractTest.java @@ -69,10 +69,9 @@ class ServletJsonContractTest { try (var context = new SpringApplicationBuilder(Application.class) .properties( - "server.port=0", "spring.main.banner-mode=off", "spring.http.converters.preferred-json-mapper=jackson2") - .run()) { + .run("--server.port=0")) { int port = ((WebServerApplicationContext) context).getWebServer().getPort(); var response = HttpClient.newHttpClient()