mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-16 04:18:17 +08:00
fix(goal): preserve graph identity in approval snapshots
This commit is contained in:
mateaix
parent
89b7dc79e1
commit
e63c513a61
@ -552,13 +552,18 @@ public class AgentService {
|
||||
if (goalApprovalReplay != null && goalApprovalReplay.applies(captured)) {
|
||||
return Flux.using(() -> acquireTurn(conversationId), permit ->
|
||||
goalApprovalReplay.replay(captured, toolCallPayload, fresh -> {
|
||||
ChatOrigin previous = ChatOriginHolder.get();
|
||||
ChatOriginHolder.set(fresh);
|
||||
return vip.mate.agent.context.GoalContinuationContext.call(true, () ->
|
||||
invokeWithLifecycleFlux(agentId, userMessage, conversationId,
|
||||
(msg, convId) -> agent.chatWithReplayStream(msg, convId, toolCallPayload,
|
||||
requesterId != null ? requesterId : ""), StreamDelta::content));
|
||||
}), vip.mate.agent.runtime.ConversationTurnGate.Permit::close)
|
||||
.doFinally(signal -> ChatOriginHolder.clear());
|
||||
try {
|
||||
return vip.mate.agent.context.GoalContinuationContext.call(true, () ->
|
||||
invokeWithLifecycleFlux(agentId, userMessage, conversationId,
|
||||
(msg, convId) -> agent.chatWithReplayStream(msg, convId, toolCallPayload,
|
||||
requesterId != null ? requesterId : ""), StreamDelta::content));
|
||||
} finally {
|
||||
if (previous == ChatOrigin.EMPTY) ChatOriginHolder.clear();
|
||||
else ChatOriginHolder.set(previous);
|
||||
}
|
||||
}), vip.mate.agent.runtime.ConversationTurnGate.Permit::close);
|
||||
}
|
||||
return Flux.defer(() -> {
|
||||
ChatOriginHolder.set(captured);
|
||||
|
||||
@ -1296,7 +1296,7 @@ public class ToolExecutionExecutor {
|
||||
ToolExecutionGuardHelper.ApprovalRequest approval = ToolExecutionGuardHelper.handleToolApproval(
|
||||
toolCall, toolName, arguments, evaluation,
|
||||
conversationId, agentId, requesterId, approvalService, streamTracker,
|
||||
events, remaining);
|
||||
events, remaining, origin);
|
||||
toolGuardService.recordApprovalAudit(guardCtx, evaluation, approval.pendingId(), autoOutcome);
|
||||
return GuardDecision.needsApproval(approval.response(), approval.pendingId());
|
||||
}
|
||||
@ -1318,7 +1318,7 @@ public class ToolExecutionExecutor {
|
||||
String approvalResponse = ToolExecutionGuardHelper.handleToolApprovalLegacy(
|
||||
toolCall, toolName, arguments, guardResult,
|
||||
conversationId, agentId, requesterId, approvalService, streamTracker,
|
||||
events, remaining);
|
||||
events, remaining, origin);
|
||||
// Legacy path never persisted a pendingId to carry here; the value
|
||||
// is unused downstream (only the boolean awaitingApproval is read).
|
||||
return GuardDecision.needsApproval(approvalResponse, null);
|
||||
|
||||
@ -5,6 +5,8 @@ import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.ai.chat.messages.AssistantMessage;
|
||||
import vip.mate.agent.GraphEventPublisher;
|
||||
import vip.mate.agent.context.ChatOrigin;
|
||||
import vip.mate.agent.context.ChatOriginHolder;
|
||||
import vip.mate.approval.ApprovalWorkflowService;
|
||||
import vip.mate.channel.web.ChatStreamTracker;
|
||||
import vip.mate.tool.guard.model.GuardEvaluation;
|
||||
@ -42,6 +44,17 @@ public final class ToolExecutionGuardHelper {
|
||||
ApprovalWorkflowService approvalService, ChatStreamTracker streamTracker,
|
||||
List<GraphEventPublisher.GraphEvent> events,
|
||||
List<AssistantMessage.ToolCall> remainingToolCalls) {
|
||||
return handleToolApproval(toolCall, toolName, arguments, evaluation, conversationId, agentId,
|
||||
requesterId, approvalService, streamTracker, events, remainingToolCalls, ChatOriginHolder.get());
|
||||
}
|
||||
|
||||
public static ApprovalRequest handleToolApproval(
|
||||
AssistantMessage.ToolCall toolCall, String toolName, String arguments,
|
||||
GuardEvaluation evaluation, String conversationId, String agentId,
|
||||
String requesterId,
|
||||
ApprovalWorkflowService approvalService, ChatStreamTracker streamTracker,
|
||||
List<GraphEventPublisher.GraphEvent> events,
|
||||
List<AssistantMessage.ToolCall> remainingToolCalls, ChatOrigin origin) {
|
||||
|
||||
if (approvalService == null) {
|
||||
log.warn("[GuardHelper] ApprovalService not available, falling back to BLOCK for tool={}", toolName);
|
||||
@ -58,9 +71,9 @@ public final class ToolExecutionGuardHelper {
|
||||
String userId = (requesterId != null && !requesterId.isEmpty()) ? requesterId : "system";
|
||||
String reason = evaluation.summary() != null ? evaluation.summary() : "需要用户审批";
|
||||
// 使用增强版 createPending，内部自动处理 findings 增强 + DB 持久化
|
||||
String pendingId = approvalService.createPending(
|
||||
String pendingId = withOrigin(origin, () -> approvalService.createPending(
|
||||
conversationId, userId, toolName, arguments, reason,
|
||||
toolCallPayload, siblingPayload, agentId, evaluation);
|
||||
toolCallPayload, siblingPayload, agentId, evaluation));
|
||||
|
||||
// SSE 直推审批事件（增强版，包含 findings）
|
||||
if (streamTracker != null) {
|
||||
@ -101,6 +114,17 @@ public final class ToolExecutionGuardHelper {
|
||||
ApprovalWorkflowService approvalService, ChatStreamTracker streamTracker,
|
||||
List<GraphEventPublisher.GraphEvent> events,
|
||||
List<AssistantMessage.ToolCall> remainingToolCalls) {
|
||||
return handleToolApprovalLegacy(toolCall, toolName, arguments, guardResult, conversationId, agentId,
|
||||
requesterId, approvalService, streamTracker, events, remainingToolCalls, ChatOriginHolder.get());
|
||||
}
|
||||
|
||||
public static String handleToolApprovalLegacy(
|
||||
AssistantMessage.ToolCall toolCall, String toolName, String arguments,
|
||||
ToolGuardResult guardResult, String conversationId, String agentId,
|
||||
String requesterId,
|
||||
ApprovalWorkflowService approvalService, ChatStreamTracker streamTracker,
|
||||
List<GraphEventPublisher.GraphEvent> events,
|
||||
List<AssistantMessage.ToolCall> remainingToolCalls, ChatOrigin origin) {
|
||||
|
||||
if (approvalService == null) {
|
||||
log.warn("[GuardHelper] ApprovalService not available, falling back to BLOCK for tool={}", toolName);
|
||||
@ -112,9 +136,9 @@ public final class ToolExecutionGuardHelper {
|
||||
String siblingPayload = serializeToolCalls(remainingToolCalls);
|
||||
|
||||
String userId = (requesterId != null && !requesterId.isEmpty()) ? requesterId : "system";
|
||||
String pendingId = approvalService.createPending(
|
||||
String pendingId = withOrigin(origin, () -> approvalService.createPending(
|
||||
conversationId, userId, toolName, arguments, guardResult.reason(),
|
||||
toolCallPayload, siblingPayload, agentId);
|
||||
toolCallPayload, siblingPayload, agentId));
|
||||
|
||||
if (streamTracker != null) {
|
||||
streamTracker.broadcastObject(conversationId, "tool_approval_requested", Map.of(
|
||||
@ -134,6 +158,16 @@ public final class ToolExecutionGuardHelper {
|
||||
|
||||
// ==================== 序列化工具 ====================
|
||||
|
||||
private static <T> T withOrigin(ChatOrigin origin, java.util.function.Supplier<T> action) {
|
||||
ChatOrigin previous = ChatOriginHolder.get();
|
||||
ChatOriginHolder.set(origin != null ? origin : ChatOrigin.EMPTY);
|
||||
try { return action.get(); }
|
||||
finally {
|
||||
if (previous == ChatOrigin.EMPTY) ChatOriginHolder.clear();
|
||||
else ChatOriginHolder.set(previous);
|
||||
}
|
||||
}
|
||||
|
||||
public static String serializeToolCall(AssistantMessage.ToolCall toolCall) {
|
||||
try {
|
||||
return OBJECT_MAPPER.writeValueAsString(Map.of(
|
||||
|
||||
@ -70,3 +70,5 @@ After interactive Web approval, Plan execution restores the original plan and ap
|
||||
When a background Goal settles into awaiting approval, its original attempt lease is released. Replaying that persisted identity cannot access managed artifacts or complete the Goal. The existing replay flow creates an exactly linked fresh attempt and lease for a Goal with selected JSON requirements. That attempt can also reuse still-eligible evidence. Single and two consecutive background approvals have been verified through the real ReAct/Plan runtime, retaining exact call and parent-attempt associations. Approval arriving during original settlement gets a short bounded wait only while both original leases and identities still match; expired or different owners remain rejected.
|
||||
|
||||
V200 records the exact attempt that settled into approval waiting and makes the approval-to-new-attempt association unique. Older waiting rows remain unbound and cannot be inferred into new execution authority. Replay renews its lease every 20 seconds and settles on normal completion. Cancellation, failure, or lease loss preserves an uncertain checkpoint; expiry recovery pauses for review instead of blindly replaying side effects. Because graph tool events can arrive in batches, an intermediate completion event does not make the whole replay safe to retry.
|
||||
|
||||
Approval snapshots use the identity bound to tool execution in graph state, rather than missing or unrelated ambient thread identity. Background replay restores the previous thread context immediately after constructing graph state; persisted account, Goal, and attempt associations are still rechecked during execution.
|
||||
|
||||
@ -72,3 +72,5 @@ JWT请求同时核对签名令牌的userId与当前启用账户ID。同名账户
|
||||
后台 Goal 进入待审批并结算后，原 attempt 的租约已经释放；其持久化审批身份不能再读写托管产物或完成 Goal。现有重放流会为已选定JSON要求的Goal建立一个准确关联的新 attempt，使用新租约执行已批准调用；新 attempt 也可复用仍合格的已有版本。单次及连续两次后台审批已通过真实ReAct/Plan运行时验证，每次批准都保留准确的调用和父attempt关联。批准早于原结算时，只对两侧租约仍有效且身份匹配的原attempt进行短暂有界等待；失租或不同owner仍拒绝。
|
||||
|
||||
V200 保存待审批结算对应的准确 attempt，并对审批生成的新 attempt 设置唯一关联。升级前的待审批行保持未绑定，不能推断为任何新的执行权限。重放每20秒续租，正常结束后结算；取消、异常或失租保留不确定检查点，到期恢复会暂停并要求核实，不盲目重放工具副作用。图的工具事件可能延后汇总，因此中途完成事件不等于整段执行可安全重试。
|
||||
|
||||
审批快照从工具执行所绑定的图身份创建，不借用环境线程中缺失或属于其他请求的身份。后台重放构建图状态后立即恢复原线程上下文；持久账户、Goal与attempt关联仍在执行时重新校验。
|
||||
|
||||
@ -80,10 +80,16 @@ class GoalJsonHttpRuntimeIntegrationTest {
|
||||
"false,supervised,true", "true,supervised,true", "false,supervised-recovered,true", "true,supervised-recovered,true",
|
||||
"false,supervised,false", "true,supervised,false", "false,supervised-recovered,false", "true,supervised-recovered,false",
|
||||
"false,approval,true", "true,approval,true", "false,scheduled-approval,true", "true,scheduled-approval,true",
|
||||
"false,scheduled-double-approval,true", "true,scheduled-double-approval,true"})
|
||||
"false,scheduled-double-approval,true", "true,scheduled-double-approval,true",
|
||||
"false,detached-approval,true", "true,detached-approval,true",
|
||||
"false,scheduled-detached-approval,true", "true,scheduled-detached-approval,true",
|
||||
"false,foreign-approval,true", "true,foreign-approval,true",
|
||||
"false,scheduled-foreign-approval,true", "true,scheduled-foreign-approval,true"})
|
||||
void authenticatedGoalCompletesThroughHttpOrScheduledProductionRuntime(boolean plan, String entry, boolean accepted) throws Exception {
|
||||
boolean approval = entry.endsWith("approval");
|
||||
boolean doubleApproval = entry.equals("scheduled-double-approval");
|
||||
boolean detached = entry.contains("detached");
|
||||
boolean foreign = entry.contains("foreign");
|
||||
boolean supervised = entry.startsWith("supervised");
|
||||
boolean scheduled = entry.startsWith("scheduled") || entry.equals("recovered") || supervised;
|
||||
boolean reuse = entry.equals("reuse");
|
||||
@ -180,6 +186,17 @@ class GoalJsonHttpRuntimeIntegrationTest {
|
||||
"{\"needs_planning\":true,\"steps\":[\"Produce, publish, check and complete the managed JSON report\"]}"))));
|
||||
}
|
||||
if (plan) step--;
|
||||
if (detached && step == 1) {
|
||||
// The graph already captured its origin. A provider/thread boundary must not
|
||||
// require that the original request ThreadLocal still be present at guard time.
|
||||
vip.mate.agent.context.ChatOriginHolder.clear();
|
||||
}
|
||||
if (foreign && step == 1) {
|
||||
vip.mate.agent.context.ChatOriginHolder.set(
|
||||
vip.mate.agent.context.ChatOrigin.web("foreign-conversation", "foreign-requester", 999L, null, null, -1L)
|
||||
.withAgent(999L).withExecutionAttribution(new vip.mate.agent.context.ExecutionAttribution(
|
||||
999L, "foreign-attempt", null, null, "foreign-fence")));
|
||||
}
|
||||
if (!accepted) {
|
||||
if (step == 0) return new ChatResponse(List.of(new Generation(AssistantMessage.builder().content("")
|
||||
.toolCalls(List.of(new AssistantMessage.ToolCall("read-unbound", "function", "getManagedGoalJsonSlots", "{}"))).build())));
|
||||
@ -305,7 +322,13 @@ class GoalJsonHttpRuntimeIntegrationTest {
|
||||
assertEquals(GoalStatus.ACTIVE, goals.getById(goal.getId()).getStatus());
|
||||
assertEquals(0, jdbc.queryForObject("SELECT COUNT(*) FROM mate_goal_json_artifact WHERE goal_id=?", Integer.class, goal.getId()));
|
||||
String persistedOrigin = jdbc.queryForObject("SELECT chat_origin FROM mate_tool_approval WHERE pending_id=?", String.class, pendingId);
|
||||
if (scheduled) assertEquals(run.attempt().id(), approvals.restoreChatOrigin(persistedOrigin).executionAttribution().goalAttemptId());
|
||||
assertEquals(conversation, approvals.restoreChatOrigin(persistedOrigin).conversationId());
|
||||
assertEquals(agentId, approvals.restoreChatOrigin(persistedOrigin).agentId());
|
||||
assertEquals(1L, approvals.restoreChatOrigin(persistedOrigin).workspaceId());
|
||||
if (scheduled) {
|
||||
assertNotNull(approvals.restoreChatOrigin(persistedOrigin).executionAttribution(), persistedOrigin);
|
||||
assertEquals(run.attempt().id(), approvals.restoreChatOrigin(persistedOrigin).executionAttribution().goalAttemptId());
|
||||
}
|
||||
else assertEquals(userId, approvals.restoreChatOrigin(persistedOrigin).requesterUserId());
|
||||
Long approvedPlan = plan ? jdbc.queryForObject("SELECT id FROM mate_plan WHERE conversation_id=?", Long.class, conversation) : null;
|
||||
planApprovalReplay.set(plan);
|
||||
|
||||
Loading…
Reference in New Issue
Block a user