- * 核心变化：不再阻塞线程等待审批。 + * INTERNAL — do not call mutating methods directly. Business code must go through + * {@link ApprovalWorkflowService}, which owns the DB / message-metadata / memory + * three-way state machine. This class only exposes: *
+ * Only {@code ApprovalWorkflowService} should call this. */ - public void resolve(String pendingId, String userId, String decision) { - PendingApproval pending = pendingMap.get(pendingId); - if (pending == null) { - throw new IllegalArgumentException("审批记录不存在或已过期: " + pendingId); - } - - if ("approved".equalsIgnoreCase(decision)) { - pending.setStatus("approved"); - } else { - pending.setStatus("denied"); - } - pending.setResolvedAt(Instant.now()); - pending.setResolvedBy(userId); - - log.info("[Approval] Resolved: id={}, decision={}, by={}", pendingId, decision, userId); + void removeFromMap(String pendingId) { + if (pendingId == null) return; + pendingMap.remove(pendingId); } /** - * Bulk-deny every pending approval still in {@code pending} status for this - * conversation. Used by the Stop endpoint to clear orphaned approvals so - * subsequent UI refreshes don't keep popping the "approve write_file?" - * banner forever, and so a `findPendingByConversation` lookup right after - * Stop returns null. + * INTERNAL — register a {@link PendingApproval} reconstructed from DB during JVM startup. + * Bypasses id generation and pre-existing-entry checks; the snapshot's {@code pendingId} + * must already match the DB row. Idempotent: if the same id already lives in the map + * (concurrent recovery path), the second call is logged and dropped. *
- * Returns the list of {@link PendingApproval} records that were marked
- * denied — callers typically use this list to update the corresponding
- * {@code mate_message.metadata.pendingApproval.status} entries in DB
- * (otherwise a page refresh re-hydrates ghost approvals from message
- * metadata even after the in-memory map is cleared).
+ * Only {@code ApprovalWorkflowService.recoverFromDb} should call this.
*/
- public List
- * 合并 resolve() + consumeApproved() 为单一操作，消除 race condition。
- *
- * @param pendingId 待审批 ID
- * @param userId 操作用户
- * @return 已消费的 PendingApproval（含 toolCallPayload），不存在或已处理返回 null
+ * INTERNAL — return the earliest {@code approved} pending matching the conversation +
+ * tool, WITHOUT removing it. Workflow uses this to take a snapshot before the
+ * two-phase DB / metadata write; the actual map removal happens via
+ * {@link #removeFromMap(String)} after commit.
*/
- public synchronized PendingApproval resolveAndConsume(String pendingId, String userId) {
- PendingApproval pending = pendingMap.get(pendingId);
- if (pending == null || !"pending".equals(pending.getStatus())) {
- log.warn("[Approval] resolveAndConsume: not found or not pending: id={}", pendingId);
- return null;
- }
- pending.setStatus("consumed");
- pending.setResolvedAt(Instant.now());
- pending.setResolvedBy(userId);
- pendingMap.remove(pendingId);
- log.info("[Approval] Resolved and consumed atomically: id={}, tool={}", pendingId, pending.getToolName());
- return pending;
- }
-
- // ==================== 消费（重放时调用） ====================
-
- /**
- * 消费已批准的审批记录（一次性消费）
- *
- * 验证 toolName 匹配（如果指定），防止参数替换攻击。
- * 移除记录并返回 PendingApproval 供重放。
- *
- * @param conversationId 会话 ID
- * @param toolName 要验证的工具名（null 跳过验证）
- * @return 已消费的 PendingApproval，或 null 如果无匹配
- */
- public PendingApproval consumeApproved(String conversationId, String toolName) {
- return consumeApproved(conversationId, toolName, null);
- }
-
- /**
- * 消费一条已审批的记录（带参数匹配校验，防止审批后参数替换攻击）
- */
- public PendingApproval consumeApproved(String conversationId, String toolName, String toolArguments) {
- PendingApproval target = pendingMap.values().stream()
+ PendingApproval findApprovedForConsume(String conversationId, String toolName) {
+ return pendingMap.values().stream()
.filter(p -> conversationId.equals(p.getConversationId()))
.filter(p -> "approved".equals(p.getStatus()))
.filter(p -> toolName == null || toolName.equals(p.getToolName()))
- .filter(p -> toolArguments == null || toolArguments.equals(p.getToolArguments()))
.min(Comparator.comparing(PendingApproval::getCreatedAt))
.orElse(null);
-
- if (target == null) {
- return null;
- }
-
- target.setStatus("consumed");
- pendingMap.remove(target.getPendingId());
- log.info("[Approval] Consumed approved: id={}, tool={}, conversation={}",
- target.getPendingId(), target.getToolName(), conversationId);
- return target;
- }
-
- // ==================== 取消与清理 ====================
-
- /**
- * 取消指定会话的所有 pending（用户发新消息时旧 pending 自动取消）
- *
- * @param conversationId 会话 ID
- * @param excludePendingId 排除的 pendingId（当前正在创建的，可为 null）
- */
- public void cancelStalePending(String conversationId, String excludePendingId) {
- pendingMap.values().stream()
- .filter(p -> conversationId.equals(p.getConversationId()))
- .filter(p -> "pending".equals(p.getStatus()))
- .filter(p -> !p.getPendingId().equals(excludePendingId))
- .forEach(p -> {
- p.setStatus("superseded");
- p.setResolvedAt(Instant.now());
- pendingMap.remove(p.getPendingId());
- log.info("[Approval] Cancelled stale pending: id={}", p.getPendingId());
- });
}
/**
- * 定时清理过期记录
- *
+ * Effective expiration follows {@code expireAt != null ? expireAt : createdAt + PENDING_TTL},
+ * so legacy / test rows whose {@code expireAt} column is NULL still time out. Expired
+ * rows are reconciled (DB → TIMEOUT, message metadata → DENIED) and skipped from
+ * the in-memory map. See RFC-067 §4.1.
*/
void recoverFromDb() {
try {
@@ -55,56 +99,95 @@ public class ApprovalWorkflowService implements ApplicationRunner {
);
int recovered = 0;
+ int expired = 0;
+ Instant now = Instant.now();
for (ToolApprovalEntity entity : pendingRecords) {
- // 检查是否已过期（30 分钟）
- if (entity.getCreatedAt() != null) {
- Instant createdAt = entity.getCreatedAt().atZone(ZoneId.systemDefault()).toInstant();
- if (Instant.now().minusSeconds(1800).isAfter(createdAt)) {
- // 已过期，更新 DB 状态
- entity.setStatus("TIMEOUT");
- entity.setResolvedAt(LocalDateTime.now());
- approvalMapper.updateById(entity);
- continue;
- }
+ // Defensive null handling: a row with neither createdAt nor expireAt is
+ // treated as freshly created so the next GC tick can revisit it instead
+ // of being silently lost.
+ Instant createdAt = entity.getCreatedAt() != null
+ ? entity.getCreatedAt().atZone(ZoneId.systemDefault()).toInstant()
+ : now;
+ Instant effectiveExpireAt = entity.getExpireAt() != null
+ ? entity.getExpireAt().atZone(ZoneId.systemDefault()).toInstant()
+ : createdAt.plus(ApprovalService.PENDING_TTL);
+
+ if (now.isAfter(effectiveExpireAt)) {
+ expireRecoveredRow(entity);
+ expired++;
+ continue;
}
- // 恢复到内存
- String pendingId = approvalService.createPending(
+ PendingApproval snapshot = new PendingApproval(
+ entity.getPendingId(),
entity.getConversationId(),
entity.getUserId(),
entity.getToolName(),
entity.getToolArguments(),
entity.getSummary(),
- entity.getToolCallPayload(),
- entity.getSiblingToolCalls(),
- entity.getAgentId()
- );
-
- // 修正内存中的 pendingId 以匹配 DB
- // 由于 ApprovalService.createPending 会生成新 ID，我们需要取消它并使用原始 ID
- approvalService.cancelStalePending(entity.getConversationId(), null);
- pendingId = approvalService.createPending(
- entity.getConversationId(),
- entity.getUserId(),
- entity.getToolName(),
- entity.getToolArguments(),
- entity.getSummary(),
- entity.getToolCallPayload(),
- entity.getSiblingToolCalls(),
- entity.getAgentId()
+ createdAt,
+ "pending"
);
+ snapshot.setToolCallPayload(entity.getToolCallPayload());
+ snapshot.setSiblingToolCalls(entity.getSiblingToolCalls());
+ snapshot.setAgentId(entity.getAgentId());
+ snapshot.setChannelType(entity.getChannelType());
+ snapshot.setRequesterName(entity.getRequesterName());
+ snapshot.setReplyTarget(entity.getReplyTarget());
+ snapshot.setFindingsJson(entity.getFindingsJson());
+ snapshot.setMaxSeverity(entity.getMaxSeverity());
+ snapshot.setSummary(entity.getSummary());
+ approvalService.registerRecovered(snapshot);
recovered++;
}
- if (recovered > 0) {
- log.info("[ApprovalWorkflow] Recovered {} pending approvals from DB", recovered);
+ if (recovered > 0 || expired > 0) {
+ log.info("[ApprovalWorkflow] DB recovery: recovered={}, expired={}", recovered, expired);
}
} catch (Exception e) {
log.warn("[ApprovalWorkflow] Failed to recover from DB (table may not exist yet): {}", e.getMessage());
}
}
+ /**
+ * Move an expired DB row to TIMEOUT and reconcile message metadata so the UI does
+ * not hydrate a ghost approval after restart.
+ *
+ * Order matters: metadata writes are gated on DB success. If {@code updateById}
+ * throws or affects zero rows, we skip the metadata flip so the three persistence
+ * loci (DB / message metadata / in-memory map) cannot drift apart — DB stuck on
+ * PENDING + metadata flipped to DENIED is the worst-case ghost state because the
+ * next recoverFromDb would re-revive the approval while the UI insists it was
+ * already settled.
+ */
+ private void expireRecoveredRow(ToolApprovalEntity entity) {
+ int rowsUpdated;
+ try {
+ entity.setStatus("TIMEOUT");
+ entity.setResolvedAt(LocalDateTime.now());
+ rowsUpdated = approvalMapper.updateById(entity);
+ } catch (Exception e) {
+ log.warn("[ApprovalWorkflow] Failed to mark expired row {} as TIMEOUT: {}",
+ entity.getPendingId(), e.getMessage());
+ return;
+ }
+ if (rowsUpdated == 0) {
+ log.warn("[ApprovalWorkflow] Expire skipped: DB row for pending {} affected 0 rows " +
+ "(concurrent resolve?); leaving metadata untouched", entity.getPendingId());
+ return;
+ }
+ try {
+ conversationService.markPendingApprovalsResolved(
+ entity.getConversationId(),
+ Set.of(entity.getPendingId()),
+ MetadataDecision.DENIED);
+ } catch (Exception e) {
+ log.warn("[ApprovalWorkflow] Failed to reconcile metadata for expired pending {}: {}",
+ entity.getPendingId(), e.getMessage());
+ }
+ }
+
/**
* 创建待审批记录（增强版，含 GuardEvaluation）
*/
@@ -144,50 +227,277 @@ public class ApprovalWorkflowService implements ApplicationRunner {
}
/**
- * 解决审批
+ * Resolve a pending approval (approve / deny) following the RFC-067 §4.2 two-phase
+ * contract: snapshot → DB UPDATE conditional on {@code status='PENDING'} →
+ * metadata reconciliation → memory mutation queued for after-commit.
+ *
+ * Idempotent under concurrent resolve: when the DB UPDATE affects 0 rows (because
+ * another caller — IM channel, GC, recoverFromDb — already moved the row off
+ * PENDING), this method returns {@link ResolveOutcome#alreadyResolved(String)}
+ * without touching metadata or in-memory state. Callers should treat this as a
+ * silent no-op; do not surface a user-facing error.
+ *
+ * On DB / metadata write failure the transaction rolls back and in-memory state
+ * stays untouched, so a retry from the next GC tick can recover. Memory mutation
+ * is registered as an {@code afterCommit} synchronization, never inline, so a
+ * post-update commit failure cannot leave memory ahead of DB.
+ *
+ * @param pendingId target approval id
+ * @param userId actor performing the resolution (for audit)
+ * @param decision case-insensitive {@code "approved"} or {@code "denied"}
+ * @return {@link ResolveOutcome} carrying the resolved snapshot + DB / metadata
+ * counters; idempotent return on no-op
*/
- public void resolve(String pendingId, String userId, String decision) {
- approvalService.resolve(pendingId, userId, decision);
- updateDbStatus(pendingId, decision.toUpperCase(), userId);
+ @Transactional
+ public ResolveOutcome resolve(String pendingId, String userId, String decision) {
+ boolean approved = "approved".equalsIgnoreCase(decision);
+ String dbStatus = approved ? "APPROVED" : "DENIED";
+ MetadataDecision metaDecision = approved ? MetadataDecision.APPROVED : MetadataDecision.DENIED;
+ String snapshotStatus = approved ? "approved" : "denied";
+
+ return performResolve(pendingId, userId, dbStatus, metaDecision, snapshotStatus,
+ /* removeFromMap */ false);
}
/**
- * 原子解决+消费
+ * Atomically resolve {@code approved} and consume the snapshot for replay.
+ * Same two-phase contract as {@link #resolve}, additionally removing the
+ * pending entry from the in-memory map after commit so a subsequent
+ * {@link #findPendingByConversation(String)} returns null and consume is
+ * single-shot. The returned {@link ResolveOutcome#consumedSnapshot()} carries
+ * {@code toolCallPayload} for replay.
*/
- public PendingApproval resolveAndConsume(String pendingId, String userId) {
- PendingApproval consumed = approvalService.resolveAndConsume(pendingId, userId);
- if (consumed != null) {
- updateDbStatus(pendingId, "CONSUMED", userId);
+ @Transactional
+ public ResolveOutcome resolveAndConsume(String pendingId, String userId) {
+ return performResolve(pendingId, userId, "CONSUMED", MetadataDecision.APPROVED,
+ "consumed", /* removeFromMap */ true);
+ }
+
+ /**
+ * Consume the earliest already-{@code approved} record for the conversation +
+ * tool — used when an out-of-band approval (e.g. /approve text command flow that
+ * resolved the record) needs to be redeemed for replay.
+ */
+ @Transactional
+ public ResolveOutcome consumeApproved(String conversationId, String toolName) {
+ PendingApproval target = approvalService.findApprovedForConsume(conversationId, toolName);
+ if (target == null) {
+ return ResolveOutcome.alreadyResolved(null);
}
- return consumed;
+ return performResolveOnSnapshot(target, null, "CONSUMED", MetadataDecision.APPROVED,
+ "consumed", /* removeFromMap */ true);
}
/**
- * 消费已批准记录
+ * Bulk-deny every pending approval in the conversation (RFC-067 §4.4.1). Used by
+ * the Web Stop endpoint to clear orphaned approvals when the user halts a turn
+ * mid-stream; without this sweep, in-flight pendings linger in the map and
+ * resurrect via metadata after refresh / restart.
+ *
+ * Two-phase per row: DB → {@code DENIED}, message metadata → {@code DENIED},
+ * map removed. Per-row failures are logged and the sweep continues; the returned
+ * list contains only the outcomes that successfully advanced through DB.
+ *
+ * @return outcome per pending that successfully transitioned to {@code DENIED}
*/
- public PendingApproval consumeApproved(String conversationId, String toolName) {
- PendingApproval consumed = approvalService.consumeApproved(conversationId, toolName);
- if (consumed != null) {
- updateDbStatus(consumed.getPendingId(), "CONSUMED", null);
+ @Transactional
+ public List
+ * The frontend's {@code Message.status} union (mateclaw-ui/src/types/index.ts) only supports
+ * {@code generating | completed | stopped | failed | awaiting_approval | interrupted}, so
+ * approval decisions never appear at the message-status layer. {@code PendingApprovalMeta.status}
+ * holds the decision ({@code approved} / {@code denied}); when the message itself was
+ * persisted as {@code awaiting_approval} we collapse it back to one of the existing terminal
+ * message states ({@code completed} for approved, {@code stopped} for denied) so downstream
+ * consumers (history sanitizer, list ordering, stuck counters) keep working unchanged.
+ *
+ * Timeout / superseded both map to {@link #DENIED} at the metadata layer (DB layer keeps
+ * the more specific {@code TIMEOUT} / {@code SUPERSEDED} status for audit purposes).
+ */
+public enum MetadataDecision {
+
+ APPROVED("approved", "completed"),
+ DENIED("denied", "stopped");
+
+ /** Target value for {@code metadata.pendingApproval.status}. */
+ public final String pendingApprovalStatus;
+
+ /** Target value for {@code MessageEntity.status} when source was {@code awaiting_approval}. */
+ public final String messageStatus;
+
+ MetadataDecision(String pendingApprovalStatus, String messageStatus) {
+ this.pendingApprovalStatus = pendingApprovalStatus;
+ this.messageStatus = messageStatus;
+ }
+}
diff --git a/mateclaw-server/src/main/java/vip/mate/approval/PendingApproval.java b/mateclaw-server/src/main/java/vip/mate/approval/PendingApproval.java
index 1bf168d8..87714e00 100644
--- a/mateclaw-server/src/main/java/vip/mate/approval/PendingApproval.java
+++ b/mateclaw-server/src/main/java/vip/mate/approval/PendingApproval.java
@@ -71,6 +71,24 @@ public class PendingApproval {
this.status = "pending";
}
+ /**
+ * INTERNAL — recovery constructor for {@code ApprovalWorkflowService.recoverFromDb}.
+ * Preserves the persisted {@code createdAt} and {@code status} so TTL/GC keep working
+ * across JVM restarts. Do not use from business paths.
+ */
+ PendingApproval(String pendingId, String conversationId, String userId,
+ String toolName, String toolArguments, String reason,
+ Instant createdAt, String status) {
+ this.pendingId = pendingId;
+ this.conversationId = conversationId;
+ this.userId = userId;
+ this.toolName = toolName;
+ this.toolArguments = toolArguments;
+ this.reason = reason;
+ this.createdAt = createdAt;
+ this.status = status;
+ }
+
// === Getters ===
public String getPendingId() { return pendingId; }
diff --git a/mateclaw-server/src/main/java/vip/mate/approval/ResolveOutcome.java b/mateclaw-server/src/main/java/vip/mate/approval/ResolveOutcome.java
new file mode 100644
index 00000000..346ff6a0
--- /dev/null
+++ b/mateclaw-server/src/main/java/vip/mate/approval/ResolveOutcome.java
@@ -0,0 +1,79 @@
+package vip.mate.approval;
+
+/**
+ * Result of an {@link ApprovalWorkflowService} state-change operation.
+ *
+ * Returned by {@code resolve}, {@code resolveAndConsume}, {@code consumeApproved},
+ * {@code cancelStalePending}, and (PR 3) {@code denyAllByConversation} so the caller can:
+ *
- * Idempotent: messages without a matching pendingApproval, or whose status
- * was already moved off {@code pending_approval}, are left untouched.
+ * For each assistant message in the conversation whose
+ * {@code metadata.pendingApproval.pendingId} appears in {@code resolvedPendingIds}
+ * and whose {@code metadata.pendingApproval.status == "pending_approval"}, this
+ * method updates three fields atomically (within a single transaction):
+ *
+ * Idempotent: messages whose metadata does not match, or whose status already moved
+ * off {@code pending_approval}, are left untouched. Timeout / superseded callers
+ * pass {@link MetadataDecision#DENIED}; the more specific terminal status lives
+ * on {@code mate_tool_approval.status} for audit (see RFC-067 §4.4.1).
*
* @param conversationId target conversation
- * @param resolvedPendingIds pendingIds whose owning message metadata should
- * flip {@code pendingApproval.status} to {@code denied}
- * @return how many messages were rewritten
+ * @param resolvedPendingIds pendingIds whose owning message metadata should be reconciled
+ * @param decision the metadata-layer decision to apply
+ * @return number of messages whose state was rewritten
*/
@Transactional
public int markPendingApprovalsResolved(String conversationId,
java.util.Set
- *
+ * INTERNAL — return a list snapshot of every {@code pending} record in the
+ * conversation, optionally excluding one id. Read-only; no map mutation.
+ * Workflow iterates this list and runs the two-phase resolve on each.
*/
- public void garbageCollect() {
- Instant now = Instant.now();
- int expiredPending = 0;
- int expiredResolved = 0;
-
- List
+ *
+ * Each pending entry's transition runs in its own transaction (markTimeout is
+ * @Transactional) so a single bad row doesn't block the rest of the sweep.
+ */
+ public void garbageCollect() {
+ Instant now = Instant.now();
+
+ // Phase A — TTL-expired pending. Snapshot first so we don't mutate a map
+ // we're iterating; markTimeout handles its own DB+metadata+memory contract.
+ int timedOut = 0;
+ for (PendingApproval expired : approvalService.snapshotExpiredPending(now)) {
+ try {
+ ResolveOutcome outcome = markTimeout(expired.getPendingId());
+ if (outcome.dbSynced()) timedOut++;
+ } catch (Exception e) {
+ log.warn("[ApprovalWorkflow] GC: markTimeout failed for {}: {}",
+ expired.getPendingId(), e.getMessage());
+ }
+ }
+
+ // Phase B — pending overflow eviction. Same path; just driven by a count cap.
+ int evictedPending = 0;
+ for (PendingApproval excess : approvalService.snapshotExcessPending(ApprovalService.MAX_PENDING)) {
+ try {
+ ResolveOutcome outcome = markTimeout(excess.getPendingId());
+ if (outcome.dbSynced()) evictedPending++;
+ } catch (Exception e) {
+ log.warn("[ApprovalWorkflow] GC: overflow markTimeout failed for {}: {}",
+ excess.getPendingId(), e.getMessage());
+ }
+ }
+
+ // Phase C — resolved cleanup. Memory-only; DB rows for these entries are
+ // already terminal (CONSUMED / DENIED / TIMEOUT / SUPERSEDED) so nothing
+ // would change in DB or metadata.
+ int droppedResolved = approvalService.dropResolvedExceedingLimits(now);
+
+ if (timedOut > 0 || evictedPending > 0 || droppedResolved > 0) {
+ log.info("[ApprovalWorkflow] GC: timed-out {}, evicted-pending {}, dropped-resolved {}, remaining={}",
+ timedOut, evictedPending, droppedResolved, approvalService.size());
+ }
+ }
+
+ // ---------- shared two-phase machinery ----------
+
+ private ResolveOutcome performResolve(String pendingId, String userId,
+ String dbStatus, MetadataDecision metaDecision,
+ String snapshotStatus, boolean removeFromMap) {
+ PendingApproval snapshot = approvalService.getPending(pendingId).orElse(null);
+ if (snapshot == null || !"pending".equals(snapshot.getStatus())) {
+ log.debug("[ApprovalWorkflow] resolve {}: not pending (snapshot={}, status={})",
+ pendingId, snapshot != null, snapshot != null ? snapshot.getStatus() : "n/a");
+ return ResolveOutcome.alreadyResolved(pendingId);
+ }
+ return performResolveOnSnapshot(snapshot, userId, dbStatus, metaDecision,
+ snapshotStatus, removeFromMap);
+ }
+
+ private ResolveOutcome performResolveOnSnapshot(PendingApproval snapshot, String userId,
+ String dbStatus, MetadataDecision metaDecision,
+ String snapshotStatus, boolean removeFromMap) {
+ // Phase 1 — DB UPDATE (conditional). The eq("PENDING") guard makes the call
+ // idempotent: if another path already won, we get rows=0 and bail without
+ // touching metadata or memory.
+ int rows;
try {
- approvalMapper.update(null, new LambdaUpdateWrapper
+ *
+ *
+ * @param pendingId the approval that was operated on (always populated)
+ * @param conversationId conversation owning the approval (null on alreadyResolved
+ * when only the id was given)
+ * @param toolName tool that was awaiting approval (null on alreadyResolved)
+ * @param decision one of: {@code approved}, {@code denied}, {@code consumed},
+ * {@code superseded}, {@code timeout}, {@code already_resolved}
+ * @param consumedSnapshot the in-memory record at the moment of consume; non-null
+ * only when {@code decision == "consumed"} (used by replay)
+ * @param dbSynced {@code true} iff the DB row's status flipped successfully
+ * in this call (false on already_resolved or DB failure)
+ * @param messagesRewritten how many {@code mate_message} rows had their
+ * {@code metadata.pendingApproval.status} reconciled
+ */
+public record ResolveOutcome(
+ String pendingId,
+ String conversationId,
+ String toolName,
+ String decision,
+ PendingApproval consumedSnapshot,
+ boolean dbSynced,
+ int messagesRewritten
+) {
+
+ public static ResolveOutcome alreadyResolved(String pendingId) {
+ return new ResolveOutcome(pendingId, null, null, "already_resolved", null, false, 0);
+ }
+
+ public static ResolveOutcome resolved(PendingApproval snapshot, String decision,
+ boolean dbSynced, int messagesRewritten) {
+ return new ResolveOutcome(
+ snapshot.getPendingId(),
+ snapshot.getConversationId(),
+ snapshot.getToolName(),
+ decision,
+ null,
+ dbSynced,
+ messagesRewritten
+ );
+ }
+
+ public static ResolveOutcome consumed(PendingApproval snapshot,
+ boolean dbSynced, int messagesRewritten) {
+ return new ResolveOutcome(
+ snapshot.getPendingId(),
+ snapshot.getConversationId(),
+ snapshot.getToolName(),
+ "consumed",
+ snapshot,
+ dbSynced,
+ messagesRewritten
+ );
+ }
+
+ public boolean isAlreadyResolved() {
+ return "already_resolved".equals(decision);
+ }
+
+ public boolean isConsumed() {
+ return "consumed".equals(decision);
+ }
+}
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java b/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java
index b1834df1..6b461f05 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/ChannelMessageRouter.java
@@ -4,7 +4,8 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import vip.mate.agent.AgentService;
-import vip.mate.approval.ApprovalService;
+import vip.mate.approval.ApprovalWorkflowService;
+import vip.mate.approval.ResolveOutcome;
import vip.mate.approval.PendingApproval;
import vip.mate.channel.model.ChannelEntity;
import vip.mate.channel.notification.ApprovalNotificationService;
@@ -47,7 +48,7 @@ public class ChannelMessageRouter {
private final ConversationService conversationService;
private final ChannelService channelService;
private final ChannelSessionStore channelSessionStore;
- private final ApprovalService approvalService;
+ private final ApprovalWorkflowService approvalService;
private final ApprovalNotificationService approvalNotificationService;
private final ConversationCompletionPublisher completionPublisher;
private final TtsService ttsService;
@@ -92,7 +93,7 @@ public class ChannelMessageRouter {
ConversationService conversationService,
ChannelService channelService,
ChannelSessionStore channelSessionStore,
- ApprovalService approvalService,
+ ApprovalWorkflowService approvalService,
ApprovalNotificationService approvalNotificationService,
ConversationCompletionPublisher completionPublisher,
TtsService ttsService,
@@ -362,36 +363,40 @@ public class ChannelMessageRouter {
adapter.getChannelType(), message.getSenderId(), originalRequester);
return;
}
- // 批准：原子解决+消费审批记录（消除 resolve/consume race condition）
- PendingApproval consumed = approvalService.resolveAndConsume(
+ // Approve via IM: workflow.resolveAndConsume runs DB + metadata + memory atomically.
+ ResolveOutcome consumeOutcome = approvalService.resolveAndConsume(
pending.getPendingId(), message.getSenderId());
- if (consumed == null) {
+ if (consumeOutcome.isAlreadyResolved()) {
adapter.sendMessage(replyTarget, "⚠️ 审批记录已过期或已被处理。");
return;
}
- log.info("[{}] Approval APPROVED via IM command: pendingId={}, tool={}",
- adapter.getChannelType(), consumed.getPendingId(), consumed.getToolName());
+ PendingApproval consumed = consumeOutcome.consumedSnapshot();
+ log.info("[{}] Approval APPROVED via IM command: pendingId={}, tool={}, msgRewritten={}",
+ adapter.getChannelType(), consumed.getPendingId(), consumed.getToolName(),
+ consumeOutcome.messagesRewritten());
replayApprovedToolCall(consumed, conversationId, adapter, message, channelEntity);
return;
} else if (isDenyCommand(userText)) {
- // 拒绝 + 清理 DB 残留审批占位消息
- approvalService.resolve(pending.getPendingId(), message.getSenderId(), "denied");
+ // Deny via IM: workflow.resolve owns the full state-machine transition.
+ ResolveOutcome denyOutcome = approvalService.resolve(
+ pending.getPendingId(), message.getSenderId(), "denied");
conversationService.removeApprovalPlaceholders(conversationId);
adapter.sendMessage(replyTarget, "⛔ 已拒绝执行工具: " + pending.getToolName());
- log.info("[{}] Approval DENIED via IM command: pendingId={}, tool={}",
- adapter.getChannelType(), pending.getPendingId(), pending.getToolName());
+ log.info("[{}] Approval DENIED via IM command: pendingId={}, tool={}, msgRewritten={}",
+ adapter.getChannelType(), pending.getPendingId(), pending.getToolName(),
+ denyOutcome.messagesRewritten());
return;
} else {
- // 非审批命令但有 pending → 视为隐式拒绝 + 清理残留
+ // Non-approval message while a pending exists → treat as implicit deny.
approvalService.resolve(pending.getPendingId(), message.getSenderId(), "denied");
conversationService.removeApprovalPlaceholders(conversationId);
adapter.sendMessage(replyTarget, "⛔ 审批已取消。将继续处理您的新消息。");
log.info("[{}] Approval auto-cancelled (non-approval message): pendingId={}",
adapter.getChannelType(), pending.getPendingId());
- // 继续正常流程处理当前消息
+ // Fall through to process the new message normally.
}
}
// ======= 审批拦截层结束 =======
diff --git a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java
index e934ce85..30e92af0 100644
--- a/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java
+++ b/mateclaw-server/src/main/java/vip/mate/channel/web/ChatController.java
@@ -17,8 +17,10 @@ import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import vip.mate.common.result.R;
import vip.mate.agent.AgentService;
import vip.mate.agent.model.AgentEntity;
-import vip.mate.approval.ApprovalService;
+import vip.mate.approval.ApprovalWorkflowService;
+import vip.mate.approval.MetadataDecision;
import vip.mate.approval.PendingApproval;
+import vip.mate.approval.ResolveOutcome;
import vip.mate.memory.event.ConversationCompletionPublisher;
import vip.mate.workspace.conversation.ConversationService;
import vip.mate.workspace.conversation.model.MessageContentPart;
@@ -55,7 +57,7 @@ public class ChatController {
private final AgentService agentService;
private final ConversationService conversationService;
- private final ApprovalService approvalService;
+ private final ApprovalWorkflowService approvalService;
private final ChatStreamTracker streamTracker;
private final ObjectMapper objectMapper;
private final ConversationCompletionPublisher completionPublisher;
@@ -190,24 +192,20 @@ public class ChatController {
return emitter;
}
- // deny: 解决并清理 DB 残留
+ // deny: workflow.resolve handles DB + metadata + memory atomically.
if (isDenyCommand) {
- approvalService.resolve(pending.getPendingId(), username, "denied");
+ ResolveOutcome denyOutcome = approvalService.resolve(pending.getPendingId(), username, "denied");
conversationService.removeApprovalPlaceholders(conversationId);
- // Sync the persisted message metadata so a subsequent page refresh
- // doesn't re-hydrate a "pending_approval" ghost from message metadata
- // that the in-memory map already moved past.
- conversationService.markPendingApprovalsResolved(conversationId,
- java.util.Set.of(pending.getPendingId()), "denied");
- log.info("[Approval-Stream] User {} denied pending {} for conversation {}",
- username, pending.getPendingId(), conversationId);
+ log.info("[Approval-Stream] User {} denied pending {} for conversation {} (dbSynced={}, msgRewritten={})",
+ username, pending.getPendingId(), conversationId,
+ denyOutcome.dbSynced(), denyOutcome.messagesRewritten());
}
- // approve: 原子 resolveAndConsume（消除 resolve/consume race condition）
+ // approve: atomic resolveAndConsume; workflow handles DB + metadata + memory.
PendingApproval consumed = null;
if (isApprovalCommand) {
- consumed = approvalService.resolveAndConsume(pending.getPendingId(), username);
- if (consumed == null) {
+ ResolveOutcome consumeOutcome = approvalService.resolveAndConsume(pending.getPendingId(), username);
+ if (consumeOutcome.isAlreadyResolved()) {
try {
sendEvent(emitter, "error", Map.of("message", "审批记录已过期或已被处理"));
sendEvent(emitter, "done", Map.of("status", "completed"));
@@ -215,15 +213,12 @@ public class ChatController {
emitter.complete();
return emitter;
}
- // 清理 DB 中残留的审批占位消息（对齐 IM 渠道 replayApprovedToolCall）
+ consumed = consumeOutcome.consumedSnapshot();
+ // Clear residual approval placeholder messages so the LLM context for
+ // replay doesn't include "[Awaiting approval]" text artifacts.
conversationService.removeApprovalPlaceholders(conversationId);
- // Same metadata sync as the deny branch — without this, refresh
- // after approval still shows the spinner-state approval card
- // because the persisted message says status='pending_approval'.
- conversationService.markPendingApprovalsResolved(conversationId,
- java.util.Set.of(consumed.getPendingId()), "approved");
- log.info("[Approval-Stream] User {} approved pending {} for conversation {}",
- username, consumed.getPendingId(), conversationId);
+ log.info("[Approval-Stream] User {} approved pending {} for conversation {} (msgRewritten={})",
+ username, consumed.getPendingId(), conversationId, consumeOutcome.messagesRewritten());
}
final PendingApproval finalConsumed = consumed;
@@ -790,23 +785,17 @@ public class ChatController {
}
boolean stopped = streamTracker.requestStop(conversationId);
- // Sweep ghost approvals — see method-level Javadoc for rationale.
- java.util.List
+ *
+ * Without this synchronization, a page refresh re-hydrates the stale
+ * {@code pending_approval} status from message metadata and the UI pops a ghost
+ * approval banner for an approval the user already settled. See RFC-067 §4.1.5.
+ *