From f6b4f7e402e1d1818d845b00758aefef97d397d3 Mon Sep 17 00:00:00 2001 From: matevip Date: Sun, 10 May 2026 08:28:20 +0800 Subject: [PATCH] feat(channel): forward async tool results to IM channels + slack file upload --- .../channel/AsyncTaskMediaDispatcher.java | 117 ++++++++++ .../channel/slack/SlackChannelAdapter.java | 204 ++++++++++++++++++ .../tool/image/ImageGenerationService.java | 38 +++- .../model3d/Model3dGenerationService.java | 25 ++- .../tool/music/MusicGenerationService.java | 54 ++++- .../tool/video/VideoGenerationService.java | 25 ++- 6 files changed, 448 insertions(+), 15 deletions(-) create mode 100644 mateclaw-server/src/main/java/vip/mate/channel/AsyncTaskMediaDispatcher.java diff --git a/mateclaw-server/src/main/java/vip/mate/channel/AsyncTaskMediaDispatcher.java b/mateclaw-server/src/main/java/vip/mate/channel/AsyncTaskMediaDispatcher.java new file mode 100644 index 00000000..d0ea75f3 --- /dev/null +++ b/mateclaw-server/src/main/java/vip/mate/channel/AsyncTaskMediaDispatcher.java @@ -0,0 +1,117 @@ +package vip.mate.channel; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Component; +import vip.mate.channel.model.ChannelSessionEntity; +import vip.mate.workspace.conversation.model.MessageContentPart; + +import java.util.List; +import java.util.Set; + +/** + * Forward async-task results (image / video / music / 3D) generated by tool + * pipelines to the IM channel that originated the conversation. + *

+ * Without this dispatcher, completion bytes only land in + * {@code mate_message} + a Web SSE broadcast — IM users (WeCom / DingTalk / + * Feishu / Telegram / etc.) see nothing arrive in their chat client because + * the tool pipeline doesn't know about channel adapters. This dispatcher + * closes that loop: look up the conversation's bound channel session, get + * the live adapter from {@link ChannelManager}, and call + * {@link ChannelAdapter#sendContentParts} so the same bytes ride the + * channel-native attachment protocol. + *

+ * Web / webchat conversations are intentionally skipped because their SSE + * stream already carries the result; double-dispatching would render the + * image twice. + */ +@Slf4j +@Component +@RequiredArgsConstructor +public class AsyncTaskMediaDispatcher { + + private final ChannelSessionStore channelSessionStore; + private final ChannelManager channelManager; + + /** + * Channel types that handle their own UX via SSE (no IM forward needed). + * Everything not in this set is treated as an IM channel and gets the + * generated parts pushed via the adapter. + */ + private static final Set WEB_CHANNEL_TYPES = Set.of("web", "webchat"); + + /** + * Forward generated content parts to the IM channel bound to this + * conversation, if any. + *

+ * Best-effort: missing session, missing adapter, or adapter exception + * are all logged at debug/warn and never propagate. The caller has + * already persisted the message to {@code mate_message} and broadcast + * to Web SSE before invoking this — IM forwarding is additive. + * + * @param conversationId the conversation id used by the agent (e.g. + * {@code wecom:XuZhanFu}, {@code dingtalk:cid_xxx}, + * {@code conv_xxx} for Web) + * @param parts assistant content parts to dispatch (typically a + * single image / video / audio / file part) + */ + public void forwardToImIfBound(String conversationId, List parts) { + if (conversationId == null || conversationId.isBlank() || parts == null || parts.isEmpty()) { + return; + } + + ChannelSessionEntity session = channelSessionStore.getSession(conversationId); + if (session == null) { + // Common case for Web-only conversations — the session is never + // populated because Web doesn't write to ChannelSessionStore. + log.debug("[async-forward] No channel session for conv={}, skipping IM forward", + conversationId); + return; + } + + String channelType = session.getChannelType(); + if (channelType == null || WEB_CHANNEL_TYPES.contains(channelType)) { + log.debug("[async-forward] conv={} is web-class ({}), skipping IM forward", + conversationId, channelType); + return; + } + + Long channelId = session.getChannelId(); + if (channelId == null) { + log.debug("[async-forward] conv={} session has no channelId, skipping", + conversationId); + return; + } + + ChannelAdapter adapter = channelManager.getAdapter(channelId).orElse(null); + if (adapter == null) { + log.warn("[async-forward] conv={} channelId={} has no live adapter (channel disabled?), skipping", + conversationId, channelId); + return; + } + + String targetId = session.getTargetId(); + if (targetId == null || targetId.isBlank()) { + log.warn("[async-forward] conv={} session has no targetId, skipping", + conversationId); + return; + } + + try { + adapter.sendContentParts(targetId, parts); + log.info("[async-forward] Dispatched {} part(s) to {} adapter (conv={}, target={})", + parts.size(), channelType, conversationId, targetId); + } catch (UnsupportedOperationException uoe) { + // Adapter doesn't override sendContentParts — fall through to + // text-only fallback. Most adapters that handle media (wecom / + // feishu / dingtalk) override; the rest will stay text-only + // until they implement the part dispatcher. + log.info("[async-forward] {} adapter does not implement sendContentParts, skipping (conv={})", + channelType, conversationId); + } catch (Exception e) { + log.warn("[async-forward] Failed to dispatch to {} adapter for conv={}: {}", + channelType, conversationId, e.getMessage()); + } + } +} diff --git a/mateclaw-server/src/main/java/vip/mate/channel/slack/SlackChannelAdapter.java b/mateclaw-server/src/main/java/vip/mate/channel/slack/SlackChannelAdapter.java index eb1b2737..7634cb5b 100644 --- a/mateclaw-server/src/main/java/vip/mate/channel/slack/SlackChannelAdapter.java +++ b/mateclaw-server/src/main/java/vip/mate/channel/slack/SlackChannelAdapter.java @@ -7,6 +7,7 @@ import com.slack.api.bolt.AppConfig; import com.slack.api.bolt.socket_mode.SocketModeApp; import com.slack.api.methods.SlackApiException; import com.slack.api.methods.response.chat.ChatPostMessageResponse; +import com.slack.api.methods.response.files.FilesUploadV2Response; import com.slack.api.model.event.MessageEvent; import lombok.extern.slf4j.Slf4j; import vip.mate.channel.AbstractChannelAdapter; @@ -14,8 +15,16 @@ import vip.mate.channel.ChannelMessage; import vip.mate.channel.ChannelMessageRouter; import vip.mate.channel.ExponentialBackoff; import vip.mate.channel.model.ChannelEntity; +import vip.mate.workspace.conversation.model.MessageContentPart; import java.io.IOException; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.Duration; import java.time.LocalDateTime; import java.util.List; import java.util.Map; @@ -295,4 +304,199 @@ public class SlackChannelAdapter extends AbstractChannelAdapter { result = result.replaceAll("(?m)^#{1,6}\\s+(.+)$", "*$1*"); return result; } + + /** + * Lazily-initialised JDK HTTP client for fetching media bytes from + * fully-qualified {@code fileUrl} fields. Only used when + * {@link MessageContentPart#getPath()} isn't set. + */ + private volatile HttpClient httpClient; + + private HttpClient getHttpClient() { + HttpClient hc = httpClient; + if (hc == null) { + synchronized (this) { + hc = httpClient; + if (hc == null) { + hc = HttpClient.newBuilder() + .connectTimeout(Duration.ofSeconds(10)) + .build(); + httpClient = hc; + } + } + } + return hc; + } + + /** + * Dispatch a list of {@link MessageContentPart}s to Slack. Text parts + * fall through to the existing {@link #sendMessage} chat-post path; + * media parts (image / audio / video / file / model3d) ride + * {@code filesUploadV2} so users see a native file card with thumbnail + * preview rather than an unopenable markdown link. + *

+ * Wired by {@link vip.mate.channel.AsyncTaskMediaDispatcher} so async + * tool results (image generation / video generation / etc.) reach + * Slack channels the same way they reach WeCom / DingTalk / Feishu. + */ + @Override + public void sendContentParts(String targetId, List parts) { + if (parts == null || parts.isEmpty()) return; + + for (MessageContentPart part : parts) { + if (part == null) continue; + String type = part.getType(); + try { + switch (type == null ? "" : type) { + case "text", "thinking" -> { + String text = part.getText(); + if (text != null && !text.isBlank()) { + sendMessage(targetId, text); + } + } + case "refusal" -> { + String text = part.getText(); + if (text != null && !text.isBlank()) { + sendMessage(targetId, "⚠️ " + text); + } + } + case "image", "audio", "video", "file", "model3d" -> uploadFilePart(targetId, part); + default -> { + // Unknown part type — fall back to its text body if any. + if (part.getText() != null && !part.getText().isBlank()) { + sendMessage(targetId, part.getText()); + } + } + } + } catch (Exception e) { + log.warn("[slack] Failed to send {} part to {}: {}", type, targetId, e.getMessage()); + sendFallbackText(targetId, part); + } + } + } + + /** + * Upload a media part to Slack via {@code files.uploadV2}. Resolves + * bytes from the part's local {@code path} first (set by image / video + * / music / 3D generation services), falling back to an HTTP fetch of + * a fully-qualified {@code fileUrl}. Returns silently after sending a + * markdown fallback if no bytes are recoverable. + */ + private void uploadFilePart(String targetId, MessageContentPart part) { + byte[] bytes = resolveBytes(part); + if (bytes == null || bytes.length == 0) { + log.warn("[slack] No bytes resolvable for {} part (path={}, fileUrl={}), falling back to text", + part.getType(), part.getPath(), part.getFileUrl()); + sendFallbackText(targetId, part); + return; + } + + String botToken = getConfigString("bot_token"); + if (botToken == null || botToken.isBlank()) { + log.warn("[slack] Missing bot_token, cannot upload {} part", part.getType()); + return; + } + + String filename = part.getFileName(); + if (filename == null || filename.isBlank()) { + filename = defaultFilenameFor(part.getType()); + } + String threadTs = lookupThreadTs(targetId); + // Slack derives the MIME from the filename's extension; passing + // contentType here is unnecessary and not part of the V2 API. + final String finalFilename = filename; + final byte[] finalBytes = bytes; + try { + FilesUploadV2Response response = slack.methods(botToken).filesUploadV2(req -> { + var b = req + .channel(targetId) + .fileData(finalBytes) + .filename(finalFilename); + if (threadTs != null && !threadTs.isBlank()) { + b.threadTs(threadTs); + } + return b; + }); + if (!response.isOk()) { + log.warn("[slack] filesUploadV2 failed for {} ({}): {}", + finalFilename, finalBytes.length, response.getError()); + sendFallbackText(targetId, part); + return; + } + log.info("[slack] Uploaded {} part ({} bytes) to {}", part.getType(), finalBytes.length, targetId); + } catch (IOException | SlackApiException e) { + log.warn("[slack] filesUploadV2 error for {}: {}", finalFilename, e.getMessage()); + sendFallbackText(targetId, part); + } + } + + /** + * Read the part's bytes from disk first (paths set by the generation + * services + the WeCom inbound pipeline both populate this), or fetch + * via HTTP when only an external {@code fileUrl} is available. Returns + * null when neither path nor URL yields bytes. + */ + private byte[] resolveBytes(MessageContentPart part) { + String path = part.getPath(); + if (path != null && !path.isBlank()) { + try { + Path p = Path.of(path); + if (Files.exists(p)) { + return Files.readAllBytes(p); + } + } catch (Exception e) { + log.debug("[slack] Reading local path failed ({}): {}", path, e.getMessage()); + } + } + String url = part.getFileUrl(); + if (url != null && (url.startsWith("http://") || url.startsWith("https://"))) { + try { + HttpRequest req = HttpRequest.newBuilder() + .uri(URI.create(url)) + .timeout(Duration.ofSeconds(30)) + .GET() + .build(); + HttpResponse resp = getHttpClient().send(req, HttpResponse.BodyHandlers.ofByteArray()); + if (resp.statusCode() == 200) { + return resp.body(); + } + log.debug("[slack] HTTP fetch returned status {} for {}", resp.statusCode(), url); + } catch (Exception e) { + log.debug("[slack] HTTP fetch failed for {}: {}", url, e.getMessage()); + } + } + return null; + } + + private static String defaultFilenameFor(String type) { + return switch (type == null ? "" : type) { + case "image" -> "image.png"; + case "audio" -> "audio.mp3"; + case "video" -> "video.mp4"; + case "model3d" -> "model.glb"; + default -> "file.bin"; + }; + } + + /** Thread-aware reply: same lookup pattern as {@link #sendMessage}. */ + private String lookupThreadTs(String channelId) { + for (var entry : threadTsCache.entrySet()) { + if (entry.getKey().contains(channelId)) { + return entry.getValue(); + } + } + return null; + } + + /** Final fallback when both upload and resolve fail — keep the user + * informed instead of dropping the message silently. */ + private void sendFallbackText(String targetId, MessageContentPart part) { + String url = part.getFileUrl(); + String fileName = part.getFileName() != null ? part.getFileName() : part.getType(); + if (url != null && !url.isBlank()) { + sendMessage(targetId, "📎 " + fileName + ": " + url); + } else { + sendMessage(targetId, "[" + fileName + "]"); + } + } } diff --git a/mateclaw-server/src/main/java/vip/mate/tool/image/ImageGenerationService.java b/mateclaw-server/src/main/java/vip/mate/tool/image/ImageGenerationService.java index 2098daed..65e483ef 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/image/ImageGenerationService.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/image/ImageGenerationService.java @@ -4,6 +4,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; +import vip.mate.channel.AsyncTaskMediaDispatcher; import vip.mate.channel.web.ChatStreamTracker; import vip.mate.system.model.SystemSettingsDTO; import vip.mate.system.service.SystemSettingService; @@ -39,6 +40,14 @@ public class ImageGenerationService { private final ImageFileDownloader fileDownloader; private final ObjectMapper objectMapper; private final ChatStreamTracker streamTracker; + /** + * Forward completion to the conversation's bound IM channel adapter so + * WeCom / DingTalk / Feishu / etc. users actually receive the generated + * image as a native attachment. Web SSE handling continues unchanged + * via {@link ChatStreamTracker} — the dispatcher is additive and skips + * Web channels to avoid double-rendering. + */ + private final AsyncTaskMediaDispatcher asyncTaskMediaDispatcher; private static final String TASK_TYPE = "image_generation"; @@ -180,7 +189,15 @@ public class ImageGenerationService { MessageContentPart imagePart = MessageContentPart.image(null, servingUrl); imagePart.setFileName(localPath.getFileName().toString()); + imagePart.setStoredName(localPath.getFileName().toString()); imagePart.setContentType("image/png"); + // Set absolute disk path so IM channel adapters can read the + // bytes locally instead of round-tripping through the + // /api/v1/chat/files endpoint (which would require auth). + imagePart.setPath(localPath.toAbsolutePath().toString()); + try { + imagePart.setFileSize(java.nio.file.Files.size(localPath)); + } catch (Exception ignored) { /* size is best-effort */ } contentParts.add(imagePart); } @@ -207,6 +224,11 @@ public class ImageGenerationService { streamTracker.broadcastObject(conversationId, "async_task_completed", data); } + // Forward to the conversation's bound IM channel adapter so + // WeCom / DingTalk / Feishu etc. users receive a native attachment + // (the SSE broadcast above only reaches Web subscribers). + asyncTaskMediaDispatcher.forwardToImIfBound(conversationId, contentParts); + log.info("[ImageGen] Sync generation completed, {} image(s) saved for conversation {}", servingUrls.size(), conversationId); @@ -248,17 +270,31 @@ public class ImageGenerationService { // 保存 assistant 消息 MessageContentPart imagePart = MessageContentPart.image(null, servingUrl); imagePart.setFileName(localPath.getFileName().toString()); + imagePart.setStoredName(localPath.getFileName().toString()); imagePart.setContentType("image/png"); + // Set absolute disk path so IM channel adapters can read the + // bytes locally instead of round-tripping through the + // /api/v1/chat/files endpoint (which would require auth). + imagePart.setPath(localPath.toAbsolutePath().toString()); + try { + imagePart.setFileSize(java.nio.file.Files.size(localPath)); + } catch (Exception ignored) { /* size is best-effort */ } + List parts = List.of(imagePart); conversationService.saveMessage( task.getConversationId(), "assistant", "图片已生成完毕", - List.of(imagePart), "completed"); + parts, "completed"); // SSE 广播（使用 imageUrl 字段） asyncTaskService.broadcastTaskEvent(task, "async_task_completed", true, null, servingUrl, null); + // Forward to the conversation's bound IM channel adapter so + // WeCom / DingTalk / Feishu etc. users receive a native + // attachment (the SSE broadcast above only reaches Web). + asyncTaskMediaDispatcher.forwardToImIfBound(task.getConversationId(), parts); + log.info("[ImageGen] Task {} completed, image saved: {}", task.getTaskId(), servingUrl); } catch (Exception e) { log.error("[ImageGen] Completion handling failed for task {}: {}", diff --git a/mateclaw-server/src/main/java/vip/mate/tool/model3d/Model3dGenerationService.java b/mateclaw-server/src/main/java/vip/mate/tool/model3d/Model3dGenerationService.java index 9fed7eb9..66d29c20 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/model3d/Model3dGenerationService.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/model3d/Model3dGenerationService.java @@ -5,6 +5,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; +import vip.mate.channel.AsyncTaskMediaDispatcher; import vip.mate.system.model.SystemSettingsDTO; import vip.mate.system.service.SystemSettingService; import vip.mate.task.AsyncTaskService; @@ -44,6 +45,12 @@ public class Model3dGenerationService { private final ConversationService conversationService; private final Model3dFileDownloader fileDownloader; private final ObjectMapper objectMapper; + /** + * Forward async-task completion to the conversation's bound IM channel + * adapter so users on WeCom / DingTalk / Feishu / etc. receive the + * generated 3D model as a native attachment. SSE remains the Web path. + */ + private final AsyncTaskMediaDispatcher asyncTaskMediaDispatcher; private static final String TASK_TYPE = "model3d_generation"; @@ -173,6 +180,7 @@ public class Model3dGenerationService { String fileName = localPath.getFileName().toString(); MessageContentPart modelPart = MessageContentPart.model3d(null, fileName); modelPart.setFileUrl(servingUrl); + modelPart.setStoredName(fileName); // model/gltf-binary for .glb is the iana-registered MIME; downstream // only cares about the URL, not the MIME header. if (fileName.endsWith(".glb")) { @@ -184,11 +192,18 @@ public class Model3dGenerationService { } else if (fileName.endsWith(".usdz")) { modelPart.setContentType("model/vnd.usdz+zip"); } + // Set absolute disk path so IM adapters can read bytes locally + // instead of round-tripping through /api/v1/chat/files (auth). + modelPart.setPath(localPath.toAbsolutePath().toString()); + try { + modelPart.setFileSize(java.nio.file.Files.size(localPath)); + } catch (Exception ignored) { /* best-effort */ } + List parts = List.of(modelPart); conversationService.saveMessage( task.getConversationId(), "assistant", "3D 模型已生成完毕", - List.of(modelPart), "completed"); + parts, "completed"); Map extra = new LinkedHashMap<>(); extra.put("modelUrl", servingUrl); @@ -196,6 +211,14 @@ public class Model3dGenerationService { asyncTaskService.broadcastTaskEventWithData(task, "async_task_completed", true, extra, null); + // Forward to the conversation's bound IM channel adapter so + // WeCom / DingTalk / Feishu etc. users receive the model as a + // native attachment (the SSE broadcast above only reaches Web). + // Most IM channels will fall back to a markdown link via + // sendFallbackText if their adapter doesn't natively support + // model/* media — that's fine, the dispatcher logs and continues. + asyncTaskMediaDispatcher.forwardToImIfBound(task.getConversationId(), parts); + log.info("[Model3dGen] Task {} completed, model saved: {}", task.getTaskId(), servingUrl); } catch (Exception e) { log.error("[Model3dGen] Completion handling failed for task {}: {}", diff --git a/mateclaw-server/src/main/java/vip/mate/tool/music/MusicGenerationService.java b/mateclaw-server/src/main/java/vip/mate/tool/music/MusicGenerationService.java index 66ebc617..ad72b750 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/music/MusicGenerationService.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/music/MusicGenerationService.java @@ -5,6 +5,7 @@ import com.fasterxml.jackson.databind.node.ObjectNode; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; +import vip.mate.channel.AsyncTaskMediaDispatcher; import vip.mate.system.model.SystemSettingsDTO; import vip.mate.system.service.SystemSettingService; import vip.mate.task.AsyncTaskService; @@ -48,6 +49,12 @@ public class MusicGenerationService { private final AsyncTaskService asyncTaskService; private final ConversationService conversationService; private final ObjectMapper objectMapper; + /** + * Forward async-task completion to the conversation's bound IM channel + * adapter so WeCom / DingTalk / Feishu / etc. users receive the generated + * audio as a native attachment. Web-class channels keep using SSE only. + */ + private final AsyncTaskMediaDispatcher asyncTaskMediaDispatcher; private static final Path UPLOAD_ROOT = Paths.get("data", "chat-uploads"); private static final String TASK_TYPE = "music_generation"; @@ -130,9 +137,10 @@ public class MusicGenerationService { return; } - String audioUrl = persistAudio(conversationId, task.getTaskId(), result); + PersistedAudio persisted = persistAudio(conversationId, task.getTaskId(), result); + String audioUrl = persisted.servingUrl(); - saveAssistantMessage(conversationId, audioUrl, result); + List parts = saveAssistantMessage(conversationId, persisted, result); ObjectNode resultJson = objectMapper.createObjectNode(); resultJson.put("audioUrl", audioUrl); @@ -152,6 +160,11 @@ public class MusicGenerationService { asyncTaskService.broadcastTaskEventWithData(task, "async_task_completed", true, extra, null); + // Forward to the conversation's bound IM channel adapter so + // WeCom / DingTalk / Feishu etc. users receive the audio as a + // native attachment (the SSE broadcast above only reaches Web). + asyncTaskMediaDispatcher.forwardToImIfBound(conversationId, parts); + log.info("[Music] Task {} succeeded, audio at {}", task.getTaskId(), audioUrl); } catch (Exception e) { log.error("[Music] Task {} worker failed: {}", task.getTaskId(), e.getMessage(), e); @@ -167,29 +180,48 @@ public class MusicGenerationService { } } - private String persistAudio(String conversationId, String taskId, - MusicGenerationResult result) throws IOException { + /** + * Stash audio bytes on disk and surface both the absolute local path and + * the browser-servable URL so callers can hand both to the + * {@link MessageContentPart}. The local path is what IM channel adapters + * read directly (faster, no auth round-trip); the serving URL is what + * the Web bubble renders. + */ + private record PersistedAudio(Path localPath, String servingUrl, String fileName) {} + + private PersistedAudio persistAudio(String conversationId, String taskId, + MusicGenerationResult result) throws IOException { Path dir = UPLOAD_ROOT.resolve(conversationId); Files.createDirectories(dir); String fileName = "music_" + taskId + "." + result.getFormat(); Path filePath = dir.resolve(fileName); Files.write(filePath, result.getAudioData()); - return "/api/v1/chat/files/" + conversationId + "/" + fileName; + String servingUrl = "/api/v1/chat/files/" + conversationId + "/" + fileName; + return new PersistedAudio(filePath, servingUrl, fileName); } - private void saveAssistantMessage(String conversationId, String audioUrl, - MusicGenerationResult result) { - MessageContentPart audioPart = MessageContentPart.audio(null, - audioUrl.substring(audioUrl.lastIndexOf('/') + 1)); - audioPart.setFileUrl(audioUrl); + private List saveAssistantMessage(String conversationId, + PersistedAudio persisted, + MusicGenerationResult result) { + MessageContentPart audioPart = MessageContentPart.audio(null, persisted.fileName()); + audioPart.setFileUrl(persisted.servingUrl()); + audioPart.setStoredName(persisted.fileName()); audioPart.setContentType(result.getContentType()); + // Set absolute disk path so IM adapters can read bytes locally + // instead of round-tripping through /api/v1/chat/files (auth). + audioPart.setPath(persisted.localPath().toAbsolutePath().toString()); + try { + audioPart.setFileSize(Files.size(persisted.localPath())); + } catch (Exception ignored) { /* best-effort */ } StringBuilder content = new StringBuilder("音乐生成完成"); if (result.getLyrics() != null && !result.getLyrics().isBlank()) { content.append("\n\n歌词:\n").append(result.getLyrics()); } + List parts = List.of(audioPart); conversationService.saveMessage(conversationId, "assistant", - content.toString(), List.of(audioPart), "completed"); + content.toString(), parts, "completed"); + return parts; } private MusicGenerationResult generateWithFallback(MusicGenerationRequest request, diff --git a/mateclaw-server/src/main/java/vip/mate/tool/video/VideoGenerationService.java b/mateclaw-server/src/main/java/vip/mate/tool/video/VideoGenerationService.java index 4b0397b8..d9ce7753 100644 --- a/mateclaw-server/src/main/java/vip/mate/tool/video/VideoGenerationService.java +++ b/mateclaw-server/src/main/java/vip/mate/tool/video/VideoGenerationService.java @@ -4,6 +4,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; +import vip.mate.channel.AsyncTaskMediaDispatcher; import vip.mate.system.model.SystemSettingsDTO; import vip.mate.system.service.SystemSettingService; import vip.mate.task.AsyncTaskService; @@ -34,6 +35,12 @@ public class VideoGenerationService { private final ConversationService conversationService; private final VideoFileDownloader fileDownloader; private final ObjectMapper objectMapper; + /** + * Forward async-task completion to the conversation's bound IM channel + * adapter so WeCom / DingTalk / Feishu / etc. users actually receive the + * generated video as a native attachment. SSE remains the Web path. + */ + private final AsyncTaskMediaDispatcher asyncTaskMediaDispatcher; private static final String TASK_TYPE = "video_generation"; @@ -176,19 +183,33 @@ public class VideoGenerationService { String servingUrl = fileDownloader.toServingUrl(task.getConversationId(), localPath); // 保存 assistant 消息（含 video content part） - MessageContentPart videoPart = MessageContentPart.video(null, localPath.getFileName().toString()); + String videoFileName = localPath.getFileName().toString(); + MessageContentPart videoPart = MessageContentPart.video(null, videoFileName); videoPart.setFileUrl(servingUrl); + videoPart.setStoredName(videoFileName); videoPart.setContentType("video/mp4"); + // Set absolute disk path so IM adapters can read bytes locally + // instead of round-tripping through /api/v1/chat/files (auth). + videoPart.setPath(localPath.toAbsolutePath().toString()); + try { + videoPart.setFileSize(java.nio.file.Files.size(localPath)); + } catch (Exception ignored) { /* best-effort */ } + List parts = List.of(videoPart); conversationService.saveMessage( task.getConversationId(), "assistant", "视频已生成完毕", - List.of(videoPart), "completed"); + parts, "completed"); // SSE 广播 asyncTaskService.broadcastTaskEvent(task, "async_task_completed", true, servingUrl, null); + // Forward to the conversation's bound IM channel adapter so + // WeCom / DingTalk / Feishu etc. users receive the video as a + // native attachment (the SSE broadcast above only reaches Web). + asyncTaskMediaDispatcher.forwardToImIfBound(task.getConversationId(), parts); + log.info("[VideoGen] Task {} completed, video saved: {}", task.getTaskId(), servingUrl); } catch (Exception e) { log.error("[VideoGen] Completion handling failed for task {}: {}",