opensource/mateclaw
2
0
Fork 0
mirror of https://gitee.com/mateos/mateclaw.git synced 2026-11-02 14:11:49 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(agent): continue unfinished work after finalization review

Browse Source
This commit is contained in:
matevip 2026-10-15 04:31:14 -04:00
parent 2efd519904
commit da67c5e8b9
7 changed files with 175 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
mateclaw-server/src/main/java/vip/mate/agent/graph/guard/TaskFinalizationCoordinator.java
View File

@ -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."}

6
mateclaw-server/src/main/java/vip/mate/agent/graph/guard/TaskFinalizationGate.java
View File

@ -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); }

10
mateclaw-server/src/test/java/vip/mate/agent/graph/guard/TaskFinalizationCoordinatorTest.java
View File

@ -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)));

13
mateclaw-server/src/test/java/vip/mate/agent/graph/guard/TaskFinalizationGateTest.java
View File

@ -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());

137
mateclaw-server/src/test/java/vip/mate/agent/graph/node/PayloadWireIntegrationTest.java Normal file
View File

@ -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<byte[]>());
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<Message> 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<String,Object>();
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<bodies.get(0).length,"serialized UTF-8 request must shrink");
var json=new ObjectMapper();
for(byte[] body:bodies) {
JsonNode messages=json.readTree(body).get("messages");
boolean hasObjective=false,hasSkill=false;
Set<String> 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); }
}
}

9
mateclaw-server/src/test/java/vip/mate/config/SchedulingConfigPausedShutdownTest.java
View File

@ -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");
}
}

3
mateclaw-server/src/test/java/vip/mate/config/ServletJsonContractTest.java
View File

@ -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()