+ * Only {@code ApprovalWorkflowService} should call this. + * + * @return number of entries removed + */ + int removeAllByConversation(String conversationId) { + if (conversationId == null) return 0; + int removed = 0; + var iter = pendingMap.entrySet().iterator(); + while (iter.hasNext()) { + var entry = iter.next(); + if (conversationId.equals(entry.getValue().getConversationId())) { + iter.remove(); + removed++; + } + } + return removed; + } + /** * INTERNAL — register a {@link PendingApproval} reconstructed from DB during JVM startup. * Bypasses id generation and pre-existing-entry checks; the snapshot's {@code pendingId} diff --git a/mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java b/mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java index 8a664032..3f9d342d 100644 --- a/mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java +++ b/mateclaw-server/src/main/java/vip/mate/approval/ApprovalWorkflowService.java @@ -10,6 +10,7 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.boot.ApplicationArguments; import org.springframework.boot.ApplicationRunner; +import org.springframework.context.event.EventListener; import org.springframework.core.annotation.Order; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; @@ -22,6 +23,7 @@ import vip.mate.approval.repository.ToolApprovalMapper; import vip.mate.tool.guard.model.GuardEvaluation; import vip.mate.tool.guard.model.GuardFinding; import vip.mate.workspace.conversation.ConversationService; +import vip.mate.workspace.conversation.event.ConversationDeletedEvent; import java.time.Instant; import java.time.LocalDateTime; @@ -82,6 +84,26 @@ public class ApprovalWorkflowService implements ApplicationRunner { } } + /** + * Drop in-memory approval state for a deleted conversation. The cascade in + * {@link ConversationService#deleteConversation} already removed the + * {@code mate_tool_approval} rows; this listener clears the parallel + * {@code pendingMap} entries so {@code findPendingByConversation} cannot + * keep returning a ghost approval that points at a non-existent + * conversation row. + *
+ * Runs after the DB cascade commits — see + * {@link ConversationDeletedEvent}. + */ + @EventListener + public void onConversationDeleted(ConversationDeletedEvent event) { + int removed = approvalService.removeAllByConversation(event.conversationId()); + if (removed > 0) { + log.info("[ApprovalWorkflow] Dropped {} in-memory pending entries for deleted conversation {}", + removed, event.conversationId()); + } + } + /** * Reconstruct in-memory pending approvals from DB at startup, preserving the * original {@code pendingId} and {@code createdAt} so subsequent resolve / GC diff --git a/mateclaw-server/src/main/java/vip/mate/workspace/conversation/ConversationService.java b/mateclaw-server/src/main/java/vip/mate/workspace/conversation/ConversationService.java index aabf3520..fea4f499 100644 --- a/mateclaw-server/src/main/java/vip/mate/workspace/conversation/ConversationService.java +++ b/mateclaw-server/src/main/java/vip/mate/workspace/conversation/ConversationService.java @@ -6,6 +6,7 @@ import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.support.TransactionSynchronization; @@ -20,6 +21,7 @@ import vip.mate.channel.model.ChannelSessionEntity; import vip.mate.channel.repository.ChannelSessionMapper; import vip.mate.task.model.AsyncTaskEntity; import vip.mate.task.repository.AsyncTaskMapper; +import vip.mate.workspace.conversation.event.ConversationDeletedEvent; import vip.mate.workspace.conversation.model.ConversationEntity; import vip.mate.workspace.conversation.model.MessageContentPart; import vip.mate.workspace.conversation.model.MessageEntity; @@ -61,6 +63,7 @@ public class ConversationService { private final ToolApprovalMapper toolApprovalMapper; private final AsyncTaskMapper asyncTaskMapper; private final ChannelSessionMapper channelSessionMapper; + private final ApplicationEventPublisher eventPublisher; /** * 获取用户的会话列表（返回 VO，包含 agentName/agentIcon/status） @@ -481,19 +484,29 @@ public class ConversationService { conversationId, messages, approvals, asyncTasks, channelSessions, childrenUnlinked, conversations); - registerAttachmentCleanupAfterCommit(conversationId); + registerPostCommitCleanup(conversationId); } - private void registerAttachmentCleanupAfterCommit(String conversationId) { + /** + * After-commit cleanup: file IO and the {@link ConversationDeletedEvent} + * fan-out both run only if the cascade actually persists, and an IO + * failure cannot roll back the DB cascade. The event lets approval and + * async-task modules drop their in-memory state (pendingMap, active + * pollers, canceled-conv set) so workers cannot resurrect orphan rows + * after the conversation row is gone. + */ + private void registerPostCommitCleanup(String conversationId) { if (TransactionSynchronizationManager.isSynchronizationActive()) { TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() { @Override public void afterCommit() { cleanAttachmentFiles(conversationId); + eventPublisher.publishEvent(new ConversationDeletedEvent(conversationId)); } }); } else { cleanAttachmentFiles(conversationId); + eventPublisher.publishEvent(new ConversationDeletedEvent(conversationId)); } } diff --git a/mateclaw-server/src/main/java/vip/mate/workspace/conversation/event/ConversationDeletedEvent.java b/mateclaw-server/src/main/java/vip/mate/workspace/conversation/event/ConversationDeletedEvent.java new file mode 100644 index 00000000..ccfc6cbb --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/workspace/conversation/event/ConversationDeletedEvent.java @@ -0,0 +1,17 @@ +package vip.mate.workspace.conversation.event; + +/** + * Fired AFTER {@link vip.mate.workspace.conversation.ConversationService#deleteConversation} + * commits its DB cascade. + *
+ * Subscribers must use this to clean up any in-memory or scheduled state keyed + * on the deleted conversation — e.g. the approval pending map, async-task + * pollers, SSE buffers, anything that survives independently of the DB row. + *
+ * Published from a {@code TransactionSynchronization.afterCommit} hook so that + * a listener observing this event can safely assume the conversation row, + * its messages, and every cascaded associate row are gone. If the transaction + * rolls back, the event is never published. + */ +public record ConversationDeletedEvent(String conversationId) { +}