fix(agent): wait for user input before action continuation (#668)

This commit is contained in:
matevip 2026-10-22 22:59:00 -04:00
parent c9c08b86fe
commit ccf33f8357
3 changed files with 247 additions and 9 deletions

View File

@ -2,22 +2,86 @@ package vip.mate.agent.graph.guard;
import vip.mate.agent.graph.state.ActionExecutionLedger;
import java.util.Optional;
import java.util.regex.Pattern;
/** Pure completion decision for action-required ReAct turns. */
public final class ActionCompletionPolicy {
public static final int MAX_RETRIES = 1;
public enum Decision { ALLOW, RETRY, UNVERIFIED, FAILED }
public enum Decision { ALLOW, WAIT_FOR_USER, RETRY, UNVERIFIED, FAILED }
private static final Pattern USER_QUESTION = Pattern.compile(
"^(?:需要我|要我|是否需要我|你(?:希望|想|需要)|您(?:希望|想|需要)|请问|"
+ "(?:能否|可以)(?:请)?(?:你|您)|"
+ "would you like\\b|do you want\\b|should I\\b|may I\\b|can I\\b|"
+ "(?:could|can|would) you\\b|"
+ "(?:which|what)\\b.*\\b(?:would you|do you|should I)\\b)",
Pattern.CASE_INSENSITIVE);
private static final Pattern USER_INPUT_REQUEST = Pattern.compile(
"^(?:请(?:确认|提供|选择|补充|告知)|please\\s+(?:provide|confirm|choose|select|specify)\\b)",
Pattern.CASE_INSENSITIVE);
private static final Pattern SOURCE_APPENDIX = Pattern.compile(
"^(?:#{1,6}\\h*)?(?:\\*\\*|__)?"
+ "(?:(?:(?:已加载)?技能的?|查询|参考|信息)?来源(?:信息)?|sources?|references?)"
+ "(?:\\*\\*|__)?\\h*[::]?\\h*$",
Pattern.CASE_INSENSITIVE | Pattern.MULTILINE);
private ActionCompletionPolicy() {
}
public static Decision evaluate(boolean actionRequired, int retryCount,
ActionExecutionLedger ledger) {
return evaluate(actionRequired, retryCount, ledger, null);
}
public static Decision evaluate(boolean actionRequired, int retryCount,
ActionExecutionLedger ledger, String content) {
// Asking for input is a turn boundary, not an unsupported completion claim.
// Check before receipts: a failed call may also need clarification to proceed.
if (userInputRequest(content).isPresent()) return Decision.WAIT_FOR_USER;
if (!actionRequired) return Decision.ALLOW;
ActionExecutionLedger evidence = ledger != null ? ledger : ActionExecutionLedger.empty();
if (evidence.hasSuccessfulSubstantiveCall()) return Decision.ALLOW;
if (evidence.hasSubstantiveAttempt()) return Decision.FAILED;
return retryCount < MAX_RETRIES ? Decision.RETRY : Decision.UNVERIFIED;
}
/** Terminal input request, preserving its formatting but excluding preceding claims. */
public static Optional<String> userInputRequest(String content) {
if (content == null || content.isBlank()) return Optional.empty();
var terminalRequest = terminalUserInputRequest(content.strip());
if (terminalRequest.isPresent()) return terminalRequest;
String original = content.strip();
// Citation appendices are presentation, not a decision to proceed past
// the preceding question. Keep ordinary trailing prose in the decision.
var sources = SOURCE_APPENDIX.matcher(original);
if (sources.find()) {
original = original.substring(0, sources.start()).stripTrailing()
.replaceFirst("\\s*\\n[-*_]{3,}\\s*$", "").stripTrailing();
}
if (original.isBlank()) return Optional.empty();
return terminalUserInputRequest(original);
}
private static Optional<String> terminalUserInputRequest(String original) {
String text = original.replaceFirst("(?:\\*\\*|__|\\*)$", "").stripTrailing();
if (text.isBlank()) return Optional.empty();
int end = text.length() - 1;
char terminal = text.charAt(end);
boolean question = terminal == '?' || terminal == '?';
// Only inspect the final sentence, so quoted examples, rhetorical questions,
// and an earlier question followed by "I'll proceed" cannot bypass the gate.
int start = end;
if (question || terminal == '.' || terminal == '。' || terminal == '!'
|| terminal == '!') start--;
while (start >= 0 && "\n。!?.!?".indexOf(text.charAt(start)) < 0) start--;
String sentence = text.substring(start + 1).stripLeading()
.replaceFirst("^(?:[-+*]|\\d+\\.)\\s+", "")
.replaceFirst("^(?:\\*\\*|__|\\*)", "");
boolean requested = USER_INPUT_REQUEST.matcher(sentence).find()
|| question && USER_QUESTION.matcher(sentence).find();
return requested ? Optional.of(original.substring(start + 1).stripLeading()) : Optional.empty();
}
}

View File

@ -1507,7 +1507,17 @@ public class ReasoningNode implements NodeAction {
String content = result.text();
ActionCompletionPolicy.Decision completionDecision = ActionCompletionPolicy.evaluate(
accessor.actionCompletionRequired(), accessor.actionCompletionRetryCount(),
accessor.actionExecutionLedger());
accessor.actionExecutionLedger(), content);
boolean waitingForUser = completionDecision == ActionCompletionPolicy.Decision.WAIT_FOR_USER;
String userInputRequest = waitingForUser
? ActionCompletionPolicy.userInputRequest(content).orElseThrow() : "";
if (waitingForUser && (accessor.actionCompletionRequired()
|| accessor.actionExecutionLedger().hasSubstantiveAttempt())
&& !accessor.actionExecutionLedger().hasSuccessfulSubstantiveCall()) {
// A question must not smuggle in an unsupported success claim. Without
// execution evidence publish only the request, never force an action.
content = userInputRequest;
}
if (completionDecision == ActionCompletionPolicy.Decision.RETRY) {
log.warn("[ReasoningNode] Rejecting text-only action completion; continuing once");
UserMessage continuation = new UserMessage("""
@ -1556,7 +1566,7 @@ public class ReasoningNode implements NodeAction {
.events(buildEvents(phaseEvent, iterStartEvent))
.build();
}
if (shouldContinueLongForm(accessor.userMessage(), accessor.longFormDraft(), content,
if (!waitingForUser && shouldContinueLongForm(accessor.userMessage(), accessor.longFormDraft(), content,
accessor.iterationCount(), accessor.maxIterations())) {
String accumulatedDraft = appendLongFormChunk(accessor.longFormDraft(), content);
int written = accumulatedDraft.length();
@ -1584,12 +1594,14 @@ public class ReasoningNode implements NodeAction {
.events(buildEvents(phaseEvent, iterStartEvent))
.build();
}
boolean longFormRequest = requestedLongFormChars(accessor.userMessage()).isPresent();
boolean longFormRequest = !waitingForUser && requestedLongFormChars(accessor.userMessage()).isPresent();
String accumulatedContent = longFormRequest
? appendLongFormChunk(accessor.longFormDraft(), content)
: (content != null ? content : "");
if (bufferedCandidate) {
boolean canContinue = (accessor.maxIterations() <= 0
// Still verify any summary preceding the question. A reviewer may reject
// that summary, but cannot authorize continuing past the user-input boundary.
if (bufferedCandidate && (!waitingForUser || !content.strip().equals(userInputRequest))) {
boolean canContinue = !waitingForUser && (accessor.maxIterations() <= 0
|| accessor.iterationCount() + 1 < accessor.maxIterations())
&& hasLlmCallBudget(accessor.maxIterations(), nextLlmCallCount, 3);
var review = finalizationCoordinator.review(chatModel, conversationId,
@ -1605,8 +1617,11 @@ public class ReasoningNode implements NodeAction {
.llmCallCount(nextLlmCallCount).mergeUsage(state,result).build();
}
var outcome = review.result().outcome();
if (outcome == vip.mate.agent.graph.guard.TaskFinalizationGate.Outcome.REVISE
|| outcome == vip.mate.agent.graph.guard.TaskFinalizationGate.Outcome.CONTINUE) {
if (waitingForUser && outcome != vip.mate.agent.graph.guard.TaskFinalizationGate.Outcome.ACCEPT) {
accumulatedContent = userInputRequest;
}
if (!waitingForUser && (outcome == vip.mate.agent.graph.guard.TaskFinalizationGate.Outcome.REVISE
|| outcome == vip.mate.agent.graph.guard.TaskFinalizationGate.Outcome.CONTINUE)) {
long progress = accessor.actionExecutionLedger().receipts().values().stream()
.filter(r -> r.substantive() && r.status() == vip.mate.agent.graph.state.ActionExecutionLedger.Status.SUCCEEDED)
.count();
@ -1640,7 +1655,7 @@ public class ReasoningNode implements NodeAction {
.streamedContent("").contentStreamed(false)
.llmCallCount(nextLlmCallCount).mergeUsage(state,result).build();
}
if (outcome != vip.mate.agent.graph.guard.TaskFinalizationGate.Outcome.ACCEPT) {
if (!waitingForUser && outcome != vip.mate.agent.graph.guard.TaskFinalizationGate.Outcome.ACCEPT) {
return reasonOutput().finalAnswer(vip.mate.agent.graph.guard.TaskFinalizationCoordinator
.incompleteAnswer(review.result().reason()))
.needsToolCall(false).shouldSummarize(false).contentStreamed(false)

View File

@ -0,0 +1,159 @@
package vip.mate.agent.graph.node;
import org.junit.jupiter.api.Test;
import org.springframework.ai.chat.model.ChatModel;
import org.springframework.ai.chat.prompt.Prompt;
import reactor.core.publisher.Flux;
import vip.mate.agent.AgentToolSet;
import vip.mate.agent.GraphEventPublisher;
import vip.mate.agent.graph.NodeStreamingChatHelper;
import vip.mate.agent.graph.guard.TaskFinalizationCoordinator;
import vip.mate.agent.graph.state.ActionExecutionLedger;
import vip.mate.agent.runtime.state.RunState;
import vip.mate.channel.web.ChatStreamTracker;
import java.util.List;
import static org.junit.jupiter.api.Assertions.*;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.*;
import static vip.mate.agent.graph.state.MateClawStateKeys.*;
class ReasoningNodeUserQuestionTest {
@Test
void skillOnlyTurnEndsAtUserQuestionWithoutForcingAction() throws Exception {
for (boolean finalization : List.of(false, true)) {
assertWaits("需要我进一步查该任务的数据质量门禁(yjy_task_quality)或与其他任务对比吗?", "load_skill", finalization, false);
assertWaits("**请确认是否继续执行?**", "load_skill", finalization, false);
assertWaits("- Please provide the task ID.", "load_skill", finalization, false);
assertWaits("Which task would you like me to query?", "load_skill", finalization, false);
}
}
@Test
void completedQueryFollowUpDoesNotEnterFinalizationOrLongFormContinuation() throws Exception {
assertWaits("以上是任务 A 的执行概况。需要我进一步查数据质量门禁吗?", "query_task", true, true);
}
@Test
void failedToolCanAskForMissingInputWithoutReplacingQuestionWithFailure() throws Exception {
assertWaits("请提供任务 ID。", "failed_query", true, false);
}
@Test
void unsupportedCompletionBeforeQuestionIsNotPublishedOrContinued() throws Exception {
for (String tool : List.of("load_skill", "failed_query")) {
var model = mock(ChatModel.class);
when(model.stream(any(Prompt.class))).thenReturn(Flux.just(ReasoningNodeFinalizationTest.response("已完成删除。需要我继续吗?")));
var tracker = mock(ChatStreamTracker.class);
var tools = mock(AgentToolSet.class);
when(tools.callbacks()).thenReturn(List.of());
var node = new ReasoningNode(model, tools, null, new NodeStreamingChatHelper(tracker), null, tracker);
var values = ReasoningNodeFinalizationTest.state();
values.put(ACTION_COMPLETION_REQUIRED, true);
values.put(ACTION_EXECUTION_LEDGER, ActionExecutionLedger.fromEvents(List.of(
GraphEventPublisher.toolComplete("call", tool, "result", tool.equals("load_skill")))));
var out = node.apply(new RunState(values));
assertEquals("需要我继续吗?", out.get(FINAL_ANSWER));
assertNotEquals(true, out.get(CONTINUE_REASONING));
verify(model, times(1)).stream(any(Prompt.class));
}
}
@Test
void sourceAppendixAfterQuestionDoesNotForceAction() throws Exception {
String question = "需要我按北京时间查询当前日期时间吗?";
for (String appendix : List.of(
"\n\n---\n\n**已加载技能来源信息**\n- 技能名称:`issue-668-confirmation`\n- 允许使用的工具:`getCurrentDateTime`\n\n本轮未执行任何日期时间查询工具。",
"\n\n来源:\n[1] issue-668-confirmation/SKILL.md",
"\n\n### Sources\n- Loaded local confirmation skill",
"\n\n**技能来源信息**\n- 技能名称:issue-668-confirmation",
"\n\n已加载技能的来源信息:\n- issue-668-confirmation/SKILL.md")) {
assertWaits(question + appendix, "load_skill", true, false, question, "ACCEPT");
}
}
@Test
void querySummaryAndQuestionBeforeSourcesStillWaitWhenReviewerRequestsContinuation() throws Exception {
String question = "需要我进一步查询当前时间吗?";
String answer = "当前日期:2026-10-23。\n\n" + question
+ "\n\n查询来源:\n- getCurrentDate 返回 2026-10-23";
assertWaits(answer, "query_task", true, false, question, "CONTINUE");
}
@Test
void ordinaryAnswerKeepsExplanationBeforeFollowUpQuestion() throws Exception {
String answer = "2 + 2 = 4. Would you like another example?";
var model = mock(ChatModel.class);
when(model.stream(any(Prompt.class))).thenReturn(Flux.just(ReasoningNodeFinalizationTest.response(answer)));
var tracker = mock(ChatStreamTracker.class);
var tools = mock(AgentToolSet.class);
when(tools.callbacks()).thenReturn(List.of());
var node = new ReasoningNode(model, tools, null, new NodeStreamingChatHelper(tracker), null, tracker);
var values = ReasoningNodeFinalizationTest.state();
values.put(ACTION_COMPLETION_REQUIRED, false);
values.put(ACTION_EXECUTION_LEDGER, ActionExecutionLedger.empty());
var out = node.apply(new RunState(values));
assertEquals(answer, out.get(FINAL_ANSWER));
assertNotEquals(true, out.get(CONTINUE_REASONING));
verify(model, times(1)).stream(any(Prompt.class));
}
@Test
void rejectedSummaryKeepsQuestionWithoutFollowingReviewerContinuation() throws Exception {
for (String decision : List.of("REVISE", "CONTINUE", "invalid")) {
assertWaits("已完成删除。需要我继续吗?", "query_task", true, false,
"需要我继续吗?", decision);
}
}
@Test
void rhetoricalOrQuotedQuestionsDoNotBypassMissingActionGuard() throws Exception {
for (String text : List.of("已完成操作。", "为什么需要质量检查?", "示例:\"需要我继续吗?\"", "需要我继续吗?我现在就执行查询。", "*", "问题示例:\n> 需要我继续吗?")) {
var model = mock(ChatModel.class);
when(model.stream(any(Prompt.class))).thenReturn(Flux.just(ReasoningNodeFinalizationTest.response(text)));
var tools = mock(AgentToolSet.class);
when(tools.callbacks()).thenReturn(List.of());
var tracker = mock(ChatStreamTracker.class);
var node = new ReasoningNode(model, tools, null, new NodeStreamingChatHelper(tracker), null, tracker);
var values = ReasoningNodeFinalizationTest.state();
values.put(ACTION_COMPLETION_REQUIRED, true);
values.put(ACTION_EXECUTION_LEDGER, ActionExecutionLedger.empty());
assertEquals(true, node.apply(new RunState(values)).get(CONTINUE_REASONING), text);
}
}
private void assertWaits(String question, String toolName, boolean finalization, boolean longForm) throws Exception {
assertWaits(question, toolName, finalization, longForm, question, "ACCEPT");
}
private void assertWaits(String question, String toolName, boolean finalization, boolean longForm,
String expected, String decision) throws Exception {
var model = mock(ChatModel.class);
when(model.stream(any(Prompt.class))).thenAnswer(call -> {
String prompt = ((Prompt) call.getArgument(0)).getContents();
return Flux.just(ReasoningNodeFinalizationTest.response(prompt.contains("You are a tool-free consistency reviewer")
? "{\"decision\":\"" + decision + "\",\"taskComplete\":" + !decision.equals("CONTINUE")
+ ",\"remainingWork\":" + decision.equals("CONTINUE")
+ ",\"failedCallCount\":0,\"recoveryClaims\":[],\"reason\":\"Summary review\"}"
: question));
});
var tools = mock(AgentToolSet.class);
when(tools.callbacks()).thenReturn(List.of());
var tracker = mock(ChatStreamTracker.class);
var helper = new NodeStreamingChatHelper(tracker);
var node = new ReasoningNode(model, tools, null, helper, null, tracker);
if (finalization) node.setFinalizationCoordinator(new TaskFinalizationCoordinator(helper, tracker));
var values = ReasoningNodeFinalizationTest.state();
values.put(ACTION_COMPLETION_REQUIRED, true);
values.put(ACTION_EXECUTION_LEDGER, ActionExecutionLedger.fromEvents(List.of(
GraphEventPublisher.toolComplete("call", toolName, "result", !toolName.equals("failed_query")))));
if (longForm) values.put(USER_MESSAGE, "请根据工具结果写一篇10000字技术报告");
var out = node.apply(new RunState(values));
assertEquals(expected, out.get(FINAL_ANSWER));
assertNotEquals(true, out.get(CONTINUE_REASONING));
assertEquals(false, out.get(NEEDS_TOOL_CALL));
assertEquals("normal", out.get(FINISH_REASON));
verify(model, times(toolName.equals("query_task") && finalization ? 2 : 1)).stream(any(Prompt.class));
}
}