fix(task): cancel running async workers on conversation delete to prevent orphan writes

mateclaw-server/src/main/java/vip/mate/task/AsyncTaskService.java

@@ -6,11 +6,13 @@ 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.stereotype.Service;
 import vip.mate.channel.web.ChatStreamTracker;
 import vip.mate.task.model.AsyncTaskEntity;
 import vip.mate.task.model.AsyncTaskInfo;
 import vip.mate.task.repository.AsyncTaskMapper;
+import vip.mate.workspace.conversation.event.ConversationDeletedEvent;
 
 import jakarta.annotation.PreDestroy;
 import java.time.LocalDateTime;
@@ -48,6 +50,33 @@ public class AsyncTaskService implements ApplicationRunner {
     /** 活跃轮询任务，key = taskId */
     private final ConcurrentHashMap<String, ScheduledFuture<?>> activePolls = new ConcurrentHashMap<>();
 
+    /** Reverse mapping taskId → conversationId so a {@link ConversationDeletedEvent}
+     *  listener can cancel every poller belonging to the deleted conversation
+     *  without scanning DB (the {@code mate_async_task} rows are gone by the
+     *  time the after-commit event fires). Populated in {@link #startPolling};
+     *  cleared in {@link #cancelPolling}. */
+    private final ConcurrentHashMap<String, String> pollTaskToConv = new ConcurrentHashMap<>();
+
+    /** Conversations whose deletion has fanned out to this service. Workers
+     *  consult {@link #isConversationCanceled} before persisting anything tied
+     *  to a conversation — the music virtual-thread worker, image/video poll
+     *  completion handlers, and any future provider-level callback are
+     *  asynchronous and may finish AFTER the conversation row + attachment
+     *  directory have already been wiped. Without this gate they would
+     *  recreate the directory + a dangling {@code mate_message} row.
+     *  <p>
+     *  Value = expiry epoch-ms. Entries older than {@link #CANCEL_RETENTION_MS}
+     *  are reaped on each event and on each lookup so the map cannot grow
+     *  without bound. The retention window is comfortably longer than
+     *  {@link #MAX_POLL_DURATION_MINUTES} and the music worker's ~120s upstream
+     *  HTTP timeout, so any in-flight worker for a deleted conversation will
+     *  still see the cancel flag when it tries to write back. */
+    private final ConcurrentHashMap<String, Long> canceledConversations = new ConcurrentHashMap<>();
+
+    /** 30 minutes — covers MAX_POLL_DURATION_MINUTES (15) + music worker's
+     *  ~2 min upstream blocking call with comfortable headroom. */
+    private static final long CANCEL_RETENTION_MS = 30L * 60 * 1000;
+
     /** 每用户最多并行任务数 */
     private static final int MAX_ACTIVE_TASKS_PER_USER = 3;
 
@@ -181,6 +210,9 @@ public class AsyncTaskService implements ApplicationRunner {
         }, 3, POLL_INTERVAL_SECONDS, TimeUnit.SECONDS);
 
         activePolls.put(taskId, future);
+        if (task.getConversationId() != null) {
+            pollTaskToConv.put(taskId, task.getConversationId());
+        }
         log.info("[AsyncTask] Started polling for task {} (interval={}s, timeout={}min)",
                 taskId, POLL_INTERVAL_SECONDS, MAX_POLL_DURATION_MINUTES);
     }
@@ -190,6 +222,49 @@ public class AsyncTaskService implements ApplicationRunner {
         if (future != null) {
             future.cancel(false);
         }
+        pollTaskToConv.remove(taskId);
+    }
+
+    // ==================== Conversation-deleted fan-out ====================
+
+    /**
+     * Returns true if this conversation was deleted recently enough that any
+     * still-running async worker (music virtual-thread, image/video poll
+     * completion, …) must abort before writing a file or persisting a
+     * message — see {@link #canceledConversations}.
+     * <p>
+     * Sweeps stale entries on read so the map stays small.
+     */
+    public boolean isConversationCanceled(String conversationId) {
+        if (conversationId == null) return false;
+        sweepCanceled();
+        return canceledConversations.containsKey(conversationId);
+    }
+
+    @EventListener
+    public void onConversationDeleted(ConversationDeletedEvent event) {
+        String convId = event.conversationId();
+        if (convId == null) return;
+
+        canceledConversations.put(convId, System.currentTimeMillis() + CANCEL_RETENTION_MS);
+
+        int cancelled = 0;
+        for (Map.Entry<String, String> entry : pollTaskToConv.entrySet()) {
+            if (convId.equals(entry.getValue())) {
+                cancelPolling(entry.getKey());
+                cancelled++;
+            }
+        }
+        if (cancelled > 0) {
+            log.info("[AsyncTask] Cancelled {} active poller(s) for deleted conversation {}",
+                    cancelled, convId);
+        }
+        sweepCanceled();
+    }
+
+    private void sweepCanceled() {
+        long now = System.currentTimeMillis();
+        canceledConversations.entrySet().removeIf(e -> e.getValue() < now);
     }
 
     // ==================== 状态更新 ====================

mateclaw-server/src/main/java/vip/mate/tool/image/ImageGenerationService.java

@@ -231,6 +231,15 @@ public class ImageGenerationService {
                     return;
                 }
 
+                // Conversation deletion can race with the final poll iteration;
+                // dropping the result avoids resurrecting attachment files and
+                // a dangling mate_message row.
+                if (asyncTaskService.isConversationCanceled(task.getConversationId())) {
+                    log.info("[ImageGen] Task {} succeeded but conversation {} was deleted, dropping result",
+                            task.getTaskId(), task.getConversationId());
+                    return;
+                }
+
                 // 下载图片到本地
                 Path localPath = fileDownloader.download(imageUrl, task.getConversationId(), task.getTaskId(), 0);
                 String servingUrl = fileDownloader.toServingUrl(task.getConversationId(), localPath);

mateclaw-server/src/main/java/vip/mate/tool/music/MusicGenerationService.java

@@ -118,6 +118,16 @@ public class MusicGenerationService {
                 return;
             }
 
+            // Conversation may have been deleted while the upstream provider was
+            // blocking (~120s). Persisting now would recreate the attachment
+            // directory we just wiped and INSERT a mate_message row pointing at
+            // a non-existent conversation. Drop the result silently.
+            if (asyncTaskService.isConversationCanceled(conversationId)) {
+                log.info("[Music] Task {} succeeded but conversation {} was deleted, dropping result",
+                        task.getTaskId(), conversationId);
+                return;
+            }
+
             String audioUrl = persistAudio(conversationId, task.getTaskId(), result);
 
             saveAssistantMessage(conversationId, audioUrl, result);

mateclaw-server/src/main/java/vip/mate/tool/video/VideoGenerationService.java

@@ -161,6 +161,15 @@ public class VideoGenerationService {
                     return;
                 }
 
+                // Conversation deletion can race with the final poll iteration;
+                // dropping the result avoids resurrecting attachment files and
+                // a dangling mate_message row.
+                if (asyncTaskService.isConversationCanceled(task.getConversationId())) {
+                    log.info("[VideoGen] Task {} succeeded but conversation {} was deleted, dropping result",
+                            task.getTaskId(), task.getConversationId());
+                    return;
+                }
+
                 // 下载视频到本地
                 Path localPath = fileDownloader.download(videoUrl, task.getConversationId(), task.getTaskId());
                 String servingUrl = fileDownloader.toServingUrl(task.getConversationId(), localPath);