From 047ac54afeb03bd0503c4938a482fcc18c0b17a5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BF=AE=E9=9B=A8?= <47820304+PeterGuy326@users.noreply.github.com> Date: Tue, 14 Jul 2026 22:31:21 +0800 Subject: [PATCH] fix(connect): forward complex message payloads (#612) Remove content-shape and message-type attachment filtering, recover forwarded unknown attachments, and preserve original media across all agent backends. --- CHANGELOG.md | 1 + .../helpers/connect_attachment_prepare.go | 141 ++++++ .../connect_attachment_prepare_test.go | 75 +++ internal/helpers/connect_chatrecord.go | 405 +++++++++++++++ internal/helpers/connect_chatrecord_test.go | 185 +++++++ internal/helpers/connect_codex_appserver.go | 28 +- .../helpers/connect_codex_appserver_test.go | 9 +- internal/helpers/connect_gemini_api.go | 213 +++++++- internal/helpers/connect_gemini_api_test.go | 72 +++ internal/helpers/connect_media.go | 461 ++++++++++++++++-- internal/helpers/connect_media_test.go | 242 +++++++++ internal/helpers/connect_opencode.go | 39 +- internal/helpers/connect_opencode_test.go | 41 ++ internal/helpers/connect_qoder_stream.go | 52 ++ internal/helpers/connect_qoder_stream_test.go | 35 +- internal/helpers/connect_stream.go | 316 ++++++++---- internal/helpers/connect_stream_test.go | 126 +++++ 17 files changed, 2279 insertions(+), 162 deletions(-) create mode 100644 internal/helpers/connect_attachment_prepare.go create mode 100644 internal/helpers/connect_attachment_prepare_test.go create mode 100644 internal/helpers/connect_chatrecord.go create mode 100644 internal/helpers/connect_chatrecord_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 1dac7ee7..07124310 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,7 @@ The format is inspired by [Keep a Changelog](https://keepachangelog.com/) and th ### Fixed +- **`dev connect` forwards complex message payloads and recoverable attachments** — messages such as forwarded chat records (`chatRecord`) are no longer discarded when they do not match the CLI's built-in text, image, or file extractors; their message type and raw JSON payload are passed to every Stream-bridge backend. Attachment discovery is based on locator capabilities (`downloadCode`, `mediaId`, `fileId`, or `dentryId` + `spaceId`) rather than a message-type allowlist, so new and unknown outer, rich-text, and nested record types retain their original bytes. Nested pictures, audio, video, and files are downloaded completely, including entries degraded by DingTalk to locator-free `unknownMsgType`: the connector acknowledges the Stream callback first, then resolves their original `mediaId` / `fileId` through the message APIs. Original bytes use each backend's native attachment path where available (OpenCode file parts, Gemini inline/Files API, Codex local images, Qoder attachments) or a narrowly allowed read-only local path for CLI/custom agents. For OpenCode, large videos are converted to a full-duration 12-frame storyboard before the model call, avoiding in-process base64 expansion and out-of-memory failures while preserving the fully downloaded original for the duration of the turn. - **Cross-platform auth regression coverage** — dedicated macOS CI now runs the Darwin-only auth/keychain regression suite with race detection, Windows CI builds and tests the native DPAPI path, and recovery guidance prefers safe migration or per-profile cleanup over destructive global reset. ## [1.0.51] - 2026-07-10 diff --git a/internal/helpers/connect_attachment_prepare.go b/internal/helpers/connect_attachment_prepare.go new file mode 100644 index 00000000..d3c96584 --- /dev/null +++ b/internal/helpers/connect_attachment_prepare.go @@ -0,0 +1,141 @@ +// Copyright 2026 Alibaba Group +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package helpers + +import ( + "context" + "fmt" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + + "github.com/google/uuid" +) + +const ( + connectVideoStoryboardFrames = 12 + connectVideoStoryboardWidth = 480 + connectVideoStoryboardMax = 16 << 20 +) + +var generateConnectVideoStoryboard = buildConnectVideoStoryboard + +// prepareOpenCodeAttachments adapts media that OpenCode's file-part bridge +// cannot safely send to the provider. OpenCode expands video files into the +// request in-process (a real 267 MiB merged-forward recording caused Bun to +// OOM), and the current bridge does not submit video/* as native multimodal +// video even when the selected model supports it. A full-duration, evenly +// sampled storyboard gives the visual model the actual sequence without +// sacrificing the separately downloaded original file. +// +// The selected iDEALab model has no audio input modality. DingTalk's message +// API normally supplies a speech-recognition transcript in the recovered +// prompt, so the opaque audio bytes are omitted from OpenCode's file parts +// instead of provoking a binary-file error. +func prepareOpenCodeAttachments(ctx context.Context, prompt string, attachments []connectMediaAttachment) (string, []connectMediaAttachment) { + prepared := make([]connectMediaAttachment, 0, len(attachments)) + for _, attachment := range attachments { + mediaType := inboundMediaType(attachment.MediaType) + switch mediaType { + case "video": + storyboard, err := generateConnectVideoStoryboard(ctx, attachment.LocalPath) + if err != nil { + fmt.Fprintf(os.Stderr, "[connect][media] OpenCode 视频故事板生成失败，跳过不安全的原视频 file part: %v\n", err) + prompt += "\n（原视频已完整下载，但当前 OpenCode 无法安全提交视频，且关键帧故事板生成失败；请明确告知用户本轮未能读取视频画面。）" + continue + } + name := strings.TrimSuffix(strings.TrimSpace(attachment.FileName), filepath.Ext(attachment.FileName)) + if name == "" { + name = "转发视频" + } + proxy := connectMediaAttachment{ + LocalPath: storyboard, + FileName: name + ".storyboard.jpg", + MediaType: "image", + } + prompt = strings.ReplaceAll(prompt, attachment.LocalPath, storyboard) + prompt += "\n（原视频已完整下载；为避免 OpenCode 展开大视频导致内存溢出，已按完整时长均匀抽取 12 帧并生成故事板图片。请按从左到右、从上到下的时间顺序分析画面。）" + prepared = append(prepared, proxy) + if info, err := os.Stat(attachment.LocalPath); err == nil { + fmt.Fprintf(os.Stderr, "[connect][media] OpenCode 视频故事板已生成: 原始=%d 字节 故事板=%s\n", info.Size(), storyboard) + } + case "audio": + prompt = strings.ReplaceAll(prompt, attachment.LocalPath, "[语音原文件已完整下载，当前模型使用钉钉转写]") + prompt += "\n（语音原文件已完整下载；当前 OpenCode 模型不接收音频 file part，请优先依据上述钉钉语音转写处理。）" + default: + prepared = append(prepared, attachment) + } + } + return prompt, prepared +} + +func buildConnectVideoStoryboard(ctx context.Context, videoPath string) (string, error) { + ffmpegPath, err := exec.LookPath("ffmpeg") + if err != nil { + return "", fmt.Errorf("未安装 ffmpeg") + } + ffprobePath, err := exec.LookPath("ffprobe") + if err != nil { + return "", fmt.Errorf("未安装 ffprobe") + } + + probe := exec.CommandContext(ctx, ffprobePath, + "-v", "error", + "-show_entries", "format=duration", + "-of", "default=noprint_wrappers=1:nokey=1", + videoPath, + ) + rawDuration, err := probe.Output() + if err != nil { + return "", fmt.Errorf("ffprobe 读取视频时长失败: %w", err) + } + duration, err := strconv.ParseFloat(strings.TrimSpace(string(rawDuration)), 64) + if err != nil || duration <= 0 { + return "", fmt.Errorf("ffprobe 返回无效视频时长 %q", strings.TrimSpace(string(rawDuration))) + } + interval := duration / connectVideoStoryboardFrames + if interval < 0.5 { + interval = 0.5 + } + + dir := filepath.Join(os.TempDir(), "dws-connect-media") + if err := os.MkdirAll(dir, 0o755); err != nil { + return "", err + } + dest := filepath.Join(dir, uuid.NewString()+".storyboard.jpg") + filter := fmt.Sprintf("fps=1/%.6f,scale=%d:-2,tile=4x3:padding=4:margin=4", interval, connectVideoStoryboardWidth) + cmd := exec.CommandContext(ctx, ffmpegPath, + "-nostdin", "-hide_banner", "-loglevel", "error", "-y", + "-i", videoPath, + "-vf", filter, + "-frames:v", "1", + "-q:v", "3", + dest, + ) + if output, err := cmd.CombinedOutput(); err != nil { + _ = os.Remove(dest) + return "", fmt.Errorf("ffmpeg 生成视频故事板失败: %w (%s)", err, truncateRunes(strings.TrimSpace(string(output)), 300)) + } + info, err := os.Stat(dest) + if err != nil { + return "", fmt.Errorf("视频故事板未生成: %w", err) + } + if info.Size() <= 0 || info.Size() > connectVideoStoryboardMax { + _ = os.Remove(dest) + return "", fmt.Errorf("视频故事板大小异常: %d 字节", info.Size()) + } + return dest, nil +} diff --git a/internal/helpers/connect_attachment_prepare_test.go b/internal/helpers/connect_attachment_prepare_test.go new file mode 100644 index 00000000..23bcaa1e --- /dev/null +++ b/internal/helpers/connect_attachment_prepare_test.go @@ -0,0 +1,75 @@ +// Copyright 2026 Alibaba Group +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package helpers + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" +) + +func TestPrepareOpenCodeAttachmentsUsesStoryboardAndTranscript(t *testing.T) { + previous := generateConnectVideoStoryboard + t.Cleanup(func() { generateConnectVideoStoryboard = previous }) + storyboard := filepath.Join(t.TempDir(), "storyboard.jpg") + if err := os.WriteFile(storyboard, []byte("jpeg"), 0o600); err != nil { + t.Fatal(err) + } + generateConnectVideoStoryboard = func(_ context.Context, path string) (string, error) { + if path != "/tmp/original.mov" { + t.Fatalf("video path = %q", path) + } + return storyboard, nil + } + + prompt, attachments := prepareOpenCodeAttachments(context.Background(), + "视频路径 /tmp/original.mov；语音路径 /tmp/original.ogg；转写：你是谁？", + []connectMediaAttachment{ + {LocalPath: "/tmp/picture.jpg", FileName: "picture.jpg", MediaType: "image"}, + {LocalPath: "/tmp/original.ogg", FileName: "voice.ogg", MediaType: "audio"}, + {LocalPath: "/tmp/original.mov", FileName: "demo.mov", MediaType: "video"}, + {LocalPath: "/tmp/report.md", FileName: "report.md", MediaType: "file"}, + }, + ) + if len(attachments) != 3 { + t.Fatalf("attachments = %#v, want image + video storyboard + file (audio omitted)", attachments) + } + if attachments[0].LocalPath != "/tmp/picture.jpg" || attachments[1].LocalPath != storyboard || attachments[1].MediaType != "image" || attachments[2].LocalPath != "/tmp/report.md" { + t.Fatalf("attachments = %#v", attachments) + } + if strings.Contains(prompt, "/tmp/original.mov") || strings.Contains(prompt, "/tmp/original.ogg") { + t.Fatalf("prompt still points OpenCode at unsupported original binary: %q", prompt) + } + for _, want := range []string{storyboard, "12 帧", "钉钉转写", "你是谁？"} { + if !strings.Contains(prompt, want) { + t.Fatalf("prompt missing %q: %q", want, prompt) + } + } +} + +func TestPrepareOpenCodeAttachmentsDoesNotSubmitVideoWhenStoryboardFails(t *testing.T) { + previous := generateConnectVideoStoryboard + t.Cleanup(func() { generateConnectVideoStoryboard = previous }) + generateConnectVideoStoryboard = func(context.Context, string) (string, error) { + return "", os.ErrNotExist + } + prompt, attachments := prepareOpenCodeAttachments(context.Background(), "请看视频", []connectMediaAttachment{ + {LocalPath: "/tmp/original.mov", FileName: "demo.mov", MediaType: "video"}, + }) + if len(attachments) != 0 || !strings.Contains(prompt, "未能读取视频画面") { + t.Fatalf("prompt=%q attachments=%#v", prompt, attachments) + } +} diff --git a/internal/helpers/connect_chatrecord.go b/internal/helpers/connect_chatrecord.go new file mode 100644 index 00000000..e11fa50b --- /dev/null +++ b/internal/helpers/connect_chatrecord.go @@ -0,0 +1,405 @@ +// Copyright 2026 Alibaba Group +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package helpers + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "os" + "path/filepath" + "regexp" + "sort" + "strings" + "time" + + "github.com/google/uuid" +) + +// chatRecordLookup records which entries DingTalk erased to unknownMsgType in +// a Stream callback. The indexes line up with forwardMessages returned by the +// user-state message API, which still exposes the original message metadata. +type chatRecordLookup struct { + MsgID string + UnknownIndexes []int +} + +type chatRecordToolCall func(context.Context, string, string, map[string]any) (string, error) + +type chatRecordMessage struct { + OpenMessageID string `json:"openMessageId"` + OpenConversationID string `json:"openConversationId"` + CreateTime string `json:"createTime"` + Content string `json:"content"` + ForwardMessages []chatRecordMessage `json:"forwardMessages"` +} + +type chatRecordMessagesEnvelope struct { + Result struct { + Messages []chatRecordMessage `json:"messages"` + } `json:"result"` +} + +type chatRecordEnrichment struct { + Prompt string + Files []fileInboundInfo + MissingCount int +} + +var ( + chatRecordMediaIDPattern = regexp.MustCompile(`\(?mediaId=([^\s)]+)\)?`) + chatRecordFileIDPattern = regexp.MustCompile(`(?i)\bfileId:\s*([^\s]+)`) + chatRecordFileNamePattern = regexp.MustCompile(`(?m)^\[文件\]\s*(.*?)(?:\s+fileId:|$)`) + chatRecordDownloadHintPattern = regexp.MustCompile(`\s*注意：如需下载使用dws\s+(?:chat message download-media|drive download)命令下载\s*`) +) + +func chatRecordEntries(content interface{}) []interface{} { + m, ok := content.(map[string]interface{}) + if !ok { + return nil + } + if entries, ok := m["chatRecord"].([]interface{}); ok { + return entries + } + if raw, ok := m["chatRecord"].(string); ok { + var entries []interface{} + if json.Unmarshal([]byte(raw), &entries) == nil { + return entries + } + } + if entries, ok := m["contents"].([]interface{}); ok { + return entries + } + return nil +} + +func chatRecordUnknownIndexes(content interface{}) []int { + entries := chatRecordEntries(content) + indexes := make([]int, 0) + for i, entry := range entries { + node, ok := entry.(map[string]interface{}) + if !ok { + continue + } + for _, key := range []string{"msgType", "msgtype", "type"} { + if value, ok := node[key].(string); ok && strings.EqualFold(strings.TrimSpace(value), "unknownMsgType") { + info := parseTypedFileInbound("file", node) + if !info.hasActionable() { + indexes = append(indexes, i) + } + break + } + } + } + return indexes +} + +// recoverChatRecordUnknowns resolves only the entries that were +// unknownMsgType in the Stream callback. It deliberately runs after the +// callback returns, so user-state MCP lookups cannot delay DingTalk's ACK. +func recoverChatRecordUnknowns(ctx context.Context, lookup chatRecordLookup, call chatRecordToolCall) (chatRecordEnrichment, error) { + var enrichment chatRecordEnrichment + if strings.TrimSpace(lookup.MsgID) == "" || len(lookup.UnknownIndexes) == 0 { + return enrichment, nil + } + + raw, err := call(ctx, "im", "list_messages_by_ids", map[string]any{ + "openMsgIds": []string{strings.TrimSpace(lookup.MsgID)}, + }) + if err != nil { + return enrichment, fmt.Errorf("查询合并转发消息: %w", err) + } + envelope, err := parseChatRecordMessages(raw) + if err != nil { + return enrichment, fmt.Errorf("解析合并转发消息: %w", err) + } + var outer *chatRecordMessage + for i := range envelope.Result.Messages { + if envelope.Result.Messages[i].OpenMessageID == strings.TrimSpace(lookup.MsgID) { + outer = &envelope.Result.Messages[i] + break + } + } + if outer == nil && len(envelope.Result.Messages) == 1 { + outer = &envelope.Result.Messages[0] + } + if outer == nil { + return enrichment, fmt.Errorf("未找到外层消息 %s", strings.TrimSpace(lookup.MsgID)) + } + + indexes := uniqueValidIndexes(lookup.UnknownIndexes, len(outer.ForwardMessages)) + if len(indexes) == 0 { + return enrichment, fmt.Errorf("unknownMsgType 索引超出转发消息范围") + } + + // list_messages_by_ids keeps mediaId for images/audio but can omit fileId + // and even stamp forwarded files with the outer conversation ID. Query every + // candidate source conversation visible in the same record and match by the + // stable inner openMessageId to recover the actual fileId. + resolvedByID := make(map[string]chatRecordMessage) + needsFileLookup := false + for _, index := range indexes { + message := outer.ForwardMessages[index] + resolvedByID[message.OpenMessageID] = message + if looksLikeForwardedFile(message.Content) && chatRecordFileID(message.Content) == "" { + needsFileLookup = true + } + } + if needsFileLookup { + enrichForwardedFileLocators(ctx, outer.ForwardMessages, resolvedByID, call) + } + + lines := []string{"已通过钉钉消息接口补拉到合并转发中原先标记为 unknownMsgType 的内容："} + for _, index := range indexes { + message := outer.ForwardMessages[index] + if resolved, ok := resolvedByID[message.OpenMessageID]; ok { + message = resolved + } + lines = append(lines, fmt.Sprintf("%d. %s", index+1, humanChatRecordContent(message.Content))) + info, ok := recoveredForwardAttachment(message) + if !ok || !info.hasActionable() { + if looksLikeForwardedAttachment(message.Content) { + enrichment.MissingCount++ + } + continue + } + enrichment.Files = append(enrichment.Files, info) + } + enrichment.Prompt = strings.Join(lines, "\n") + return enrichment, nil +} + +func parseChatRecordMessages(raw string) (chatRecordMessagesEnvelope, error) { + var envelope chatRecordMessagesEnvelope + if err := json.Unmarshal([]byte(raw), &envelope); err != nil { + return envelope, err + } + return envelope, nil +} + +func uniqueValidIndexes(indexes []int, length int) []int { + seen := make(map[int]struct{}, len(indexes)) + out := make([]int, 0, len(indexes)) + for _, index := range indexes { + if index < 0 || index >= length { + continue + } + if _, ok := seen[index]; ok { + continue + } + seen[index] = struct{}{} + out = append(out, index) + } + sort.Ints(out) + return out +} + +func enrichForwardedFileLocators(ctx context.Context, all []chatRecordMessage, resolved map[string]chatRecordMessage, call chatRecordToolCall) { + candidates := make(map[string]struct{}) + var earliest time.Time + for _, message := range all { + if conversationID := strings.TrimSpace(message.OpenConversationID); conversationID != "" { + candidates[conversationID] = struct{}{} + } + if parsed, err := time.ParseInLocation("2006-01-02 15:04:05", strings.TrimSpace(message.CreateTime), time.Local); err == nil && (earliest.IsZero() || parsed.Before(earliest)) { + earliest = parsed + } + } + if earliest.IsZero() { + return + } + start := earliest.Add(-time.Minute).Format("2006-01-02 15:04:05") + for conversationID := range candidates { + raw, err := call(ctx, "chat", "list_conversation_message_v2", map[string]any{ + "openconversation_id": conversationID, + "time": start, + "forward": true, + "limit": 50, + }) + if err != nil { + continue + } + envelope, err := parseChatRecordMessages(raw) + if err != nil { + continue + } + for _, message := range envelope.Result.Messages { + current, wanted := resolved[message.OpenMessageID] + if !wanted || chatRecordFileID(message.Content) == "" { + continue + } + current.Content = message.Content + current.OpenConversationID = message.OpenConversationID + if strings.TrimSpace(current.CreateTime) == "" { + current.CreateTime = message.CreateTime + } + resolved[message.OpenMessageID] = current + } + } +} + +func recoveredForwardAttachment(message chatRecordMessage) (fileInboundInfo, bool) { + content := strings.TrimSpace(message.Content) + info := fileInboundInfo{ + OpenMessageID: strings.TrimSpace(message.OpenMessageID), + OpenConversationID: strings.TrimSpace(message.OpenConversationID), + } + if mediaID := chatRecordMediaID(content); mediaID != "" { + info.MediaID = mediaID + switch { + case strings.Contains(content, "[图片消息]"): + info.MediaType, info.FileName = "image", "转发图片" + case strings.Contains(content, "[语音消息]"): + info.MediaType, info.FileName = "audio", "转发语音.bin" + case strings.Contains(content, "[视频消息]"): + info.MediaType, info.FileName = "video", "转发视频.bin" + default: + info.MediaType, info.FileName = "file", "转发媒体.bin" + } + return info, true + } + if !looksLikeForwardedFile(content) { + return fileInboundInfo{}, false + } + info.FileID = chatRecordFileID(content) + info.FileName = chatRecordFileName(content) + if info.FileName == "" { + info.FileName = "转发文件" + } + info.MediaType = mediaTypeFromFileName(info.FileName) + return info, true +} + +func chatRecordMediaID(content string) string { + match := chatRecordMediaIDPattern.FindStringSubmatch(content) + if len(match) < 2 { + return "" + } + return strings.TrimSpace(match[1]) +} + +func chatRecordFileID(content string) string { + match := chatRecordFileIDPattern.FindStringSubmatch(content) + if len(match) < 2 { + return "" + } + return strings.TrimSpace(match[1]) +} + +func chatRecordFileName(content string) string { + match := chatRecordFileNamePattern.FindStringSubmatch(content) + if len(match) < 2 { + return "" + } + return strings.TrimSpace(match[1]) +} + +func looksLikeForwardedFile(content string) bool { + return strings.Contains(content, "[文件]") +} + +func looksLikeForwardedAttachment(content string) bool { + return looksLikeForwardedFile(content) || chatRecordMediaID(content) != "" +} + +func humanChatRecordContent(content string) string { + content = chatRecordDownloadHintPattern.ReplaceAllString(content, "") + content = chatRecordMediaIDPattern.ReplaceAllString(content, "") + content = chatRecordFileIDPattern.ReplaceAllString(content, "") + content = strings.TrimSpace(content) + if content == "" { + return "[无法提取文字内容]" + } + return content +} + +func mediaTypeFromFileName(fileName string) string { + switch strings.ToLower(filepath.Ext(strings.TrimSpace(fileName))) { + case ".mp4", ".mov", ".m4v", ".avi", ".mkv", ".webm": + return "video" + case ".mp3", ".m4a", ".aac", ".wav", ".amr", ".ogg", ".flac": + return "audio" + case ".png", ".jpg", ".jpeg", ".gif", ".webp", ".bmp", ".heic": + return "image" + default: + return "file" + } +} + +func (c *aiCardClient) downloadRecoveredChatRecordFile(ctx context.Context, info fileInboundInfo) (string, error) { + return c.downloadRecoveredChatRecordFileWithCall(ctx, info, callMCPToolReturnTextOnServer) +} + +func (c *aiCardClient) downloadRecoveredChatRecordFileWithCall(ctx context.Context, info fileInboundInfo, call chatRecordToolCall) (string, error) { + var raw string + var err error + switch { + case strings.TrimSpace(info.MediaID) != "": + raw, err = call(ctx, "im", "get_resource_download_url", map[string]any{ + "resourceType": "mediaId", + "resourceId": strings.TrimSpace(info.MediaID), + "openMessageId": strings.TrimSpace(info.OpenMessageID), + "openConversationId": strings.TrimSpace(info.OpenConversationID), + }) + case strings.TrimSpace(info.FileID) != "": + raw, err = call(ctx, "drive", "download_file", map[string]any{ + "fileId": strings.TrimSpace(info.FileID), + }) + default: + return "", fmt.Errorf("转发附件缺少 mediaId/fileId") + } + if err != nil { + return "", err + } + resourceURL, headers, err := parseDownloadInfo(raw) + if err != nil { + return "", err + } + return downloadConnectURLToTemp(ctx, c.httpClient, resourceURL, headers, info.FileName) +} + +func downloadConnectURLToTemp(ctx context.Context, client *http.Client, resourceURL string, headers map[string]string, fileName string) (string, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, resourceURL, nil) + if err != nil { + return "", err + } + for key, value := range headers { + req.Header.Set(key, value) + } + resp, err := connectMediaDownloadClient(client).Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + if resp.StatusCode >= 400 { + return "", fmt.Errorf("转发附件下载 HTTP %d", resp.StatusCode) + } + if resp.ContentLength > mediaMaxDownloadBytes { + return "", fmt.Errorf("转发附件过大：%d 字节，最大允许 %d 字节", resp.ContentLength, mediaMaxDownloadBytes) + } + dir := filepath.Join(os.TempDir(), "dws-connect-media") + if err := os.MkdirAll(dir, 0o755); err != nil { + return "", err + } + ext := filepath.Ext(filepath.Base(strings.TrimSpace(fileName))) + if ext == "" { + ext = mediaExt(resourceURL, resp.Header.Get("Content-Type")) + } + dest := filepath.Join(dir, uuid.NewString()+ext) + if err := writeCompleteMediaFile(dest, resp.Body); err != nil { + return "", err + } + return dest, nil +} diff --git a/internal/helpers/connect_chatrecord_test.go b/internal/helpers/connect_chatrecord_test.go new file mode 100644 index 00000000..2dbfcf9e --- /dev/null +++ b/internal/helpers/connect_chatrecord_test.go @@ -0,0 +1,185 @@ +// Copyright 2026 Alibaba Group +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package helpers + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "reflect" + "strings" + "testing" +) + +func TestChatRecordUnknownIndexesMatchesObservedCallback(t *testing.T) { + content := map[string]interface{}{ + "chatRecord": `[{"msgType":"picture","downloadCode":"pic-live"},{"msgType":"unknownMsgType"},{"msgType":"unknownMsgType"},{"msgType":"text","content":"[合并的聊天记录]"},{"msgType":"unknownMsgType"}]`, + } + if got, want := chatRecordUnknownIndexes(content), []int{1, 2, 4}; !reflect.DeepEqual(got, want) { + t.Fatalf("unknown indexes = %v, want %v", got, want) + } +} + +func TestRecoverChatRecordUnknownsRestoresAudioVideoAndFile(t *testing.T) { + outer := `{ + "result": {"messages": [{ + "openMessageId": "outer-1", + "openConversationId": "outer-conv", + "createTime": "2026-07-14 16:16:44", + "forwardMessages": [ + {"openMessageId":"pic-1","openConversationId":"source-conv","createTime":"2026-07-14 14:33:15","content":"[图片消息](mediaId=picture-media) 注意：如需下载使用dws chat message download-media命令下载"}, + {"openMessageId":"audio-1","openConversationId":"source-conv","createTime":"2026-07-14 14:33:19","content":"你是谁？你是谁\n[语音消息](mediaId=audio-media) 注意：如需下载使用dws chat message download-media命令下载"}, + {"openMessageId":"video-1","openConversationId":"outer-conv","createTime":"2026-07-14 14:33:36","content":"[文件] 录屏.mov"}, + {"openMessageId":"text-1","openConversationId":"source-conv","createTime":"2026-07-14 14:34:47","content":"普通文本"}, + {"openMessageId":"file-1","openConversationId":"outer-conv","createTime":"2026-07-14 14:34:57","content":"[文件] 巡检报告(1)(1).md"} + ] + }]} +}` + sourceConversation := `{ + "result": {"messages": [ + {"openMessageId":"video-1","openConversationId":"source-conv","createTime":"2026-07-14 14:33:36","content":"[文件] 录屏.mov fileId: video-file-id 注意：如需下载使用dws drive download命令下载"}, + {"openMessageId":"file-1","openConversationId":"source-conv","createTime":"2026-07-14 14:34:57","content":"[文件] 巡检报告.md fileId: markdown-file-id 注意：如需下载使用dws drive download命令下载"} + ]} +}` + + var calls []string + call := func(_ context.Context, server, tool string, args map[string]any) (string, error) { + calls = append(calls, server+"."+tool) + switch tool { + case "list_messages_by_ids": + if server != "im" || !reflect.DeepEqual(args["openMsgIds"], []string{"outer-1"}) { + t.Fatalf("list_messages_by_ids route/args = %s %#v", server, args) + } + return outer, nil + case "list_conversation_message_v2": + if server != "chat" || args["forward"] != true || args["limit"] != 50 { + t.Fatalf("list_conversation_message_v2 route/args = %s %#v", server, args) + } + if args["openconversation_id"] == "source-conv" { + return sourceConversation, nil + } + return `{"result":{"messages":[]}}`, nil + default: + return "", fmt.Errorf("unexpected tool %s", tool) + } + } + + enrichment, err := recoverChatRecordUnknowns(context.Background(), chatRecordLookup{ + MsgID: "outer-1", + UnknownIndexes: []int{1, 2, 4}, + }, call) + if err != nil { + t.Fatal(err) + } + if enrichment.MissingCount != 0 || len(enrichment.Files) != 3 { + t.Fatalf("enrichment = %#v, want 3 recovered files and no missing attachment", enrichment) + } + wants := []fileInboundInfo{ + {MediaID: "audio-media", OpenMessageID: "audio-1", OpenConversationID: "source-conv", FileName: "转发语音.bin", MediaType: "audio"}, + {FileID: "video-file-id", OpenMessageID: "video-1", OpenConversationID: "source-conv", FileName: "录屏.mov", MediaType: "video"}, + {FileID: "markdown-file-id", OpenMessageID: "file-1", OpenConversationID: "source-conv", FileName: "巡检报告.md", MediaType: "file"}, + } + for i, want := range wants { + if !reflect.DeepEqual(enrichment.Files[i], want) { + t.Fatalf("files[%d] = %#v, want %#v", i, enrichment.Files[i], want) + } + } + if !strings.Contains(enrichment.Prompt, "你是谁？你是谁") || !strings.Contains(enrichment.Prompt, "录屏.mov") || !strings.Contains(enrichment.Prompt, "巡检报告.md") { + t.Fatalf("prompt did not preserve recovered user content: %q", enrichment.Prompt) + } + if strings.Contains(enrichment.Prompt, "mediaId") || strings.Contains(enrichment.Prompt, "fileId") || strings.Contains(enrichment.Prompt, "dws drive") { + t.Fatalf("prompt leaked transport locators/instructions: %q", enrichment.Prompt) + } + if len(calls) != 3 || calls[0] != "im.list_messages_by_ids" { + t.Fatalf("calls = %v, want outer lookup plus both candidate conversations", calls) + } +} + +func TestRecoverChatRecordUnknownsKeepsMetadataWhenFileLocatorMissing(t *testing.T) { + outer := `{"result":{"messages":[{"openMessageId":"outer-1","forwardMessages":[{"openMessageId":"file-1","openConversationId":"wrong-conv","createTime":"2026-07-14 14:34:57","content":"[文件] report.md"}]}]}}` + call := func(_ context.Context, _, tool string, _ map[string]any) (string, error) { + if tool == "list_messages_by_ids" { + return outer, nil + } + return `{"result":{"messages":[]}}`, nil + } + enrichment, err := recoverChatRecordUnknowns(context.Background(), chatRecordLookup{MsgID: "outer-1", UnknownIndexes: []int{0}}, call) + if err != nil { + t.Fatal(err) + } + if enrichment.MissingCount != 1 || len(enrichment.Files) != 0 || !strings.Contains(enrichment.Prompt, "report.md") { + t.Fatalf("enrichment = %#v, want metadata prompt plus one honestly missing attachment", enrichment) + } +} + +func TestDownloadRecoveredChatRecordFileRoutesAndPreservesOriginalBytes(t *testing.T) { + wantBody := []byte("original attachment bytes") + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Header.Get("X-Signed-Test") != "yes" { + http.Error(w, "missing signed header", http.StatusForbidden) + return + } + _, _ = w.Write(wantBody) + })) + defer server.Close() + + tests := []struct { + name string + info fileInboundInfo + wantServer string + wantTool string + wantArg string + wantValue string + }{ + { + name: "mediaId audio", + info: fileInboundInfo{MediaID: "media-1", OpenMessageID: "msg-1", OpenConversationID: "conv-1", FileName: "voice.amr", MediaType: "audio"}, + wantServer: "im", wantTool: "get_resource_download_url", wantArg: "resourceId", wantValue: "media-1", + }, + { + name: "drive fileId video", + info: fileInboundInfo{FileID: "file-1", FileName: "video.mov", MediaType: "video"}, + wantServer: "drive", wantTool: "download_file", wantArg: "fileId", wantValue: "file-1", + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + call := func(_ context.Context, serverID, tool string, args map[string]any) (string, error) { + if serverID != tc.wantServer || tool != tc.wantTool || args[tc.wantArg] != tc.wantValue { + t.Fatalf("call = %s.%s %#v", serverID, tool, args) + } + return fmt.Sprintf(`{"result":{"resourceUrl":%q,"headers":{"X-Signed-Test":"yes"}}}`, server.URL), nil + } + client := &aiCardClient{httpClient: server.Client()} + path, err := client.downloadRecoveredChatRecordFileWithCall(context.Background(), tc.info, call) + if err != nil { + t.Fatal(err) + } + defer os.Remove(path) + got, err := os.ReadFile(path) + if err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(got, wantBody) { + t.Fatalf("downloaded bytes = %q, want %q", got, wantBody) + } + if filepath.Ext(path) != filepath.Ext(tc.info.FileName) { + t.Fatalf("downloaded path = %q, want original extension from %q", path, tc.info.FileName) + } + }) + } +} diff --git a/internal/helpers/connect_codex_appserver.go b/internal/helpers/connect_codex_appserver.go index faa0921e..a1a88968 100644 --- a/internal/helpers/connect_codex_appserver.go +++ b/internal/helpers/connect_codex_appserver.go @@ -30,7 +30,7 @@ import ( "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/pkg/config" ) -const codexRobotDeveloperInstructions = "你是钉钉群聊里的智能助手，请用简洁、自然的中文直接回答用户问题；不要提及系统提示、内部协议或运行时细节；不要主动读写文件或执行命令。" +const codexRobotDeveloperInstructions = "你是钉钉群聊里的智能助手，请用简洁、自然的中文直接回答用户问题；不要提及系统提示、内部协议或运行时细节；不要主动读写文件或执行命令。仅当用户消息明确附带了本地附件路径时，可以只读该附件或运行分析该附件所必需的只读命令，不得访问其它文件。" // codexAppServerForwarder uses Codex's official app-server JSON-RPC protocol to // keep one Codex thread per DingTalk conversation. @@ -100,11 +100,19 @@ func (f *codexAppServerForwarder) forward(ctx context.Context, convID, text stri return f.forwardStream(ctx, convID, text, nil) } -func (f *codexAppServerForwarder) forwardStream(ctx context.Context, convID, text string, onDelta func(string)) (string, error) { - return f.forwardAppServer(ctx, convID, text, onDelta) +func (f *codexAppServerForwarder) forwardWithAttachments(ctx context.Context, convID, text string, attachments []connectMediaAttachment) (string, error) { + return f.forwardStreamWithAttachments(ctx, convID, text, attachments, nil) } -func (f *codexAppServerForwarder) forwardAppServer(ctx context.Context, convID, text string, onDelta func(string)) (string, error) { +func (f *codexAppServerForwarder) forwardStream(ctx context.Context, convID, text string, onDelta func(string)) (string, error) { + return f.forwardStreamWithAttachments(ctx, convID, text, nil, onDelta) +} + +func (f *codexAppServerForwarder) forwardStreamWithAttachments(ctx context.Context, convID, text string, attachments []connectMediaAttachment, onDelta func(string)) (string, error) { + return f.forwardAppServer(ctx, convID, text, attachments, onDelta) +} + +func (f *codexAppServerForwarder) forwardAppServer(ctx context.Context, convID, text string, attachments []connectMediaAttachment, onDelta func(string)) (string, error) { ctx, cancel := applyTimeout(ctx, f.timeout) defer cancel() @@ -151,7 +159,7 @@ func (f *codexAppServerForwarder) forwardAppServer(ctx context.Context, convID, } } - reply, err := cli.runTurn(ctx, threadID, text, onDelta) + reply, err := cli.runTurn(ctx, threadID, text, attachments, onDelta) if err != nil { return "", err } @@ -436,13 +444,19 @@ func (c *codexAppServerClient) resumeThread(ctx context.Context, params map[stri return codexThreadIDFromResult(c.waitResponse(ctx, id)) } -func (c *codexAppServerClient) runTurn(ctx context.Context, threadID, text string, onDelta func(string)) (string, error) { +func (c *codexAppServerClient) runTurn(ctx context.Context, threadID, text string, attachments []connectMediaAttachment, onDelta func(string)) (string, error) { id := c.requestID() + input := []map[string]string{{"type": "text", "text": text}} + for _, attachment := range attachments { + if attachment.MediaType == "image" && strings.TrimSpace(attachment.LocalPath) != "" { + input = append(input, map[string]string{"type": "localImage", "path": attachment.LocalPath}) + } + } if err := c.send(map[string]any{ "id": id, "method": "turn/start", "params": map[string]any{ - "input": []map[string]string{{"type": "text", "text": text}}, + "input": input, "threadId": threadID, }, }); err != nil { diff --git a/internal/helpers/connect_codex_appserver_test.go b/internal/helpers/connect_codex_appserver_test.go index e6e7423f..6115e9da 100644 --- a/internal/helpers/connect_codex_appserver_test.go +++ b/internal/helpers/connect_codex_appserver_test.go @@ -122,9 +122,13 @@ while IFS= read -r line; do workDir: dir, sessions: newCodexThreadSessions(""), } + imagePath := filepath.Join(dir, "forwarded.png") + if err := os.WriteFile(imagePath, []byte("png-bytes"), 0o600); err != nil { + t.Fatal(err) + } var deltas []string - reply, err := fwd.forwardStream(context.Background(), "conv-1", "第一问", func(s string) { + reply, err := fwd.forwardStreamWithAttachments(context.Background(), "conv-1", "第一问", []connectMediaAttachment{{LocalPath: imagePath, FileName: "forwarded.png", MediaType: "image"}}, func(s string) { deltas = append(deltas, s) }) if err != nil { @@ -151,6 +155,9 @@ while IFS= read -r line; do if strings.Count(log, `"method":"thread/resume"`) != 1 { t.Fatalf("expected one thread/resume, log:\n%s", log) } + if !strings.Contains(log, `"type":"localImage"`) || !strings.Contains(log, imagePath) { + t.Fatalf("turn/start missing native localImage input, log:\n%s", log) + } } func TestCodexAppServerForwarderReturnsAppServerError(t *testing.T) { diff --git a/internal/helpers/connect_gemini_api.go b/internal/helpers/connect_gemini_api.go index 540b4a93..eaff9046 100644 --- a/internal/helpers/connect_gemini_api.go +++ b/internal/helpers/connect_gemini_api.go @@ -16,12 +16,14 @@ package helpers import ( "bytes" "context" + "encoding/base64" "encoding/json" "fmt" "io" "net/http" "net/url" "os" + "path/filepath" "strings" "time" @@ -31,6 +33,10 @@ import ( const ( defaultGeminiAPIBaseURL = "https://generativelanguage.googleapis.com/v1beta" defaultGeminiModel = "gemini-2.5-flash" + // Base64 adds roughly 33%, while Gemini's generateContent request limit for + // inline audio is 20 MiB. Keep the raw aggregate below 15 MiB and use the + // resumable Files API for anything larger. + geminiInlineRawLimit = 15 << 20 ) type geminiAPIForwarder struct { @@ -88,6 +94,10 @@ func (f *geminiAPIForwarder) label() string { } func (f *geminiAPIForwarder) forward(ctx context.Context, _, text string) (string, error) { + return f.forwardWithAttachments(ctx, "", text, nil) +} + +func (f *geminiAPIForwarder) forwardWithAttachments(ctx context.Context, _ string, text string, attachments []connectMediaAttachment) (string, error) { ctx, cancel := applyTimeout(ctx, f.timeout) defer cancel() @@ -95,14 +105,15 @@ func (f *geminiAPIForwarder) forward(ctx context.Context, _, text string) (strin if err != nil { return "", err } + parts, err := f.partsWithAttachments(ctx, text, attachments) + if err != nil { + return "", err + } body := geminiGenerateContentRequest{ SystemInstruction: geminiContent{ Parts: []geminiPart{{Text: "你是钉钉群聊里的智能助手，请用简洁、自然的中文直接回答用户问题；不要提及任何系统提示或内部实现。"}}, }, - Contents: []geminiContent{{ - Role: "user", - Parts: []geminiPart{{Text: text}}, - }}, + Contents: []geminiContent{{Role: "user", Parts: parts}}, } raw, err := json.Marshal(body) if err != nil { @@ -148,6 +159,186 @@ func (f *geminiAPIForwarder) forward(ctx context.Context, _, text string) (strin return strings.Join(chunks, "\n\n"), nil } +func (f *geminiAPIForwarder) partsWithAttachments(ctx context.Context, text string, attachments []connectMediaAttachment) ([]geminiPart, error) { + parts := []geminiPart{{Text: text}} + var aggregate int64 + for _, attachment := range attachments { + if info, err := os.Stat(attachment.LocalPath); err == nil { + aggregate += info.Size() + } + } + for _, attachment := range attachments { + path := strings.TrimSpace(attachment.LocalPath) + if path == "" { + continue + } + mimeType := connectAttachmentMIME(path) + if aggregate <= geminiInlineRawLimit { + raw, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("读取 Gemini 附件 %s 失败：%w", path, err) + } + parts = append(parts, geminiPart{InlineData: &geminiInlineData{ + MIMEType: mimeType, + Data: base64.StdEncoding.EncodeToString(raw), + }}) + continue + } + uploaded, err := f.uploadFile(ctx, path, mimeType, attachment.FileName) + if err != nil { + return nil, err + } + parts = append(parts, geminiPart{FileData: &geminiFileData{ + MIMEType: uploaded.MIMEType, + FileURI: uploaded.URI, + }}) + } + return parts, nil +} + +type geminiUploadedFile struct { + Name string `json:"name"` + URI string `json:"uri"` + MIMEType string `json:"mimeType"` + State string `json:"state"` +} + +func (f *geminiAPIForwarder) uploadFile(ctx context.Context, path, mimeType, displayName string) (geminiUploadedFile, error) { + info, err := os.Stat(path) + if err != nil { + return geminiUploadedFile{}, fmt.Errorf("读取 Gemini 附件信息 %s 失败：%w", path, err) + } + if strings.TrimSpace(displayName) == "" { + displayName = filepath.Base(path) + } + meta, err := json.Marshal(map[string]any{"file": map[string]any{"display_name": displayName}}) + if err != nil { + return geminiUploadedFile{}, err + } + startURL, err := f.filesEndpoint(true) + if err != nil { + return geminiUploadedFile{}, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, startURL, bytes.NewReader(meta)) + if err != nil { + return geminiUploadedFile{}, err + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("x-goog-api-key", f.apiKey) + req.Header.Set("X-Goog-Upload-Protocol", "resumable") + req.Header.Set("X-Goog-Upload-Command", "start") + req.Header.Set("X-Goog-Upload-Header-Content-Length", fmt.Sprint(info.Size())) + req.Header.Set("X-Goog-Upload-Header-Content-Type", mimeType) + resp, err := f.httpClient.Do(req) + if err != nil { + return geminiUploadedFile{}, err + } + startBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1024*1024)) + resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return geminiUploadedFile{}, fmt.Errorf("gemini Files API 启动上传 HTTP %d: %s", resp.StatusCode, truncateRunes(strings.TrimSpace(string(startBody)), 300)) + } + uploadURL := strings.TrimSpace(resp.Header.Get("X-Goog-Upload-URL")) + if uploadURL == "" { + return geminiUploadedFile{}, fmt.Errorf("gemini Files API 未返回 X-Goog-Upload-URL") + } + fh, err := os.Open(path) + if err != nil { + return geminiUploadedFile{}, err + } + defer fh.Close() + uploadReq, err := http.NewRequestWithContext(ctx, http.MethodPost, uploadURL, fh) + if err != nil { + return geminiUploadedFile{}, err + } + uploadReq.ContentLength = info.Size() + uploadReq.Header.Set("Content-Type", mimeType) + uploadReq.Header.Set("X-Goog-Upload-Offset", "0") + uploadReq.Header.Set("X-Goog-Upload-Command", "upload, finalize") + uploadResp, err := f.httpClient.Do(uploadReq) + if err != nil { + return geminiUploadedFile{}, err + } + defer uploadResp.Body.Close() + raw, _ := io.ReadAll(io.LimitReader(uploadResp.Body, 4*1024*1024)) + if uploadResp.StatusCode < 200 || uploadResp.StatusCode >= 300 { + return geminiUploadedFile{}, fmt.Errorf("gemini Files API 上传 HTTP %d: %s", uploadResp.StatusCode, truncateRunes(strings.TrimSpace(string(raw)), 300)) + } + var envelope struct { + File geminiUploadedFile `json:"file"` + } + if err := json.Unmarshal(raw, &envelope); err != nil { + return geminiUploadedFile{}, err + } + if envelope.File.URI == "" { + return geminiUploadedFile{}, fmt.Errorf("gemini Files API 上传结果缺少 file.uri") + } + if envelope.File.MIMEType == "" { + envelope.File.MIMEType = mimeType + } + return f.waitForUploadedFile(ctx, envelope.File) +} + +func (f *geminiAPIForwarder) waitForUploadedFile(ctx context.Context, file geminiUploadedFile) (geminiUploadedFile, error) { + for strings.EqualFold(file.State, "PROCESSING") { + select { + case <-ctx.Done(): + return geminiUploadedFile{}, ctx.Err() + case <-time.After(time.Second): + } + base, err := f.filesEndpoint(false) + if err != nil { + return geminiUploadedFile{}, err + } + name := strings.TrimPrefix(strings.TrimLeft(file.Name, "/"), "files/") + statusURL := strings.TrimRight(base, "/") + "/" + name + req, err := http.NewRequestWithContext(ctx, http.MethodGet, statusURL, nil) + if err != nil { + return geminiUploadedFile{}, err + } + req.Header.Set("x-goog-api-key", f.apiKey) + resp, err := f.httpClient.Do(req) + if err != nil { + return geminiUploadedFile{}, err + } + raw, _ := io.ReadAll(io.LimitReader(resp.Body, 1024*1024)) + resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + return geminiUploadedFile{}, fmt.Errorf("gemini Files API 查询 HTTP %d: %s", resp.StatusCode, truncateRunes(strings.TrimSpace(string(raw)), 300)) + } + if err := json.Unmarshal(raw, &file); err != nil { + return geminiUploadedFile{}, err + } + } + if strings.EqualFold(file.State, "FAILED") { + return geminiUploadedFile{}, fmt.Errorf("gemini Files API 处理附件失败: %s", file.Name) + } + return file, nil +} + +func (f *geminiAPIForwarder) filesEndpoint(upload bool) (string, error) { + base := strings.TrimRight(strings.TrimSpace(f.baseURL), "/") + if base == "" { + base = defaultGeminiAPIBaseURL + } + u, err := url.Parse(base) + if err != nil || u.Scheme == "" || u.Host == "" { + return "", fmt.Errorf("GEMINI_API_BASE_URL 无效") + } + path := strings.TrimRight(u.Path, "/") + if upload { + if strings.HasSuffix(path, "/v1beta") { + path = strings.TrimSuffix(path, "/v1beta") + "/upload/v1beta/files" + } else { + path += "/upload/v1beta/files" + } + } else if !strings.HasSuffix(path, "/files") { + path += "/files" + } + u.Path = path + return u.String(), nil +} + func (f *geminiAPIForwarder) generateContentEndpoint() (string, error) { base := strings.TrimRight(strings.TrimSpace(f.baseURL), "/") if base == "" { @@ -174,7 +365,19 @@ type geminiContent struct { } type geminiPart struct { - Text string `json:"text"` + Text string `json:"text,omitempty"` + InlineData *geminiInlineData `json:"inlineData,omitempty"` + FileData *geminiFileData `json:"fileData,omitempty"` +} + +type geminiInlineData struct { + MIMEType string `json:"mimeType"` + Data string `json:"data"` +} + +type geminiFileData struct { + MIMEType string `json:"mimeType"` + FileURI string `json:"fileUri"` } type geminiGenerateContentResponse struct { diff --git a/internal/helpers/connect_gemini_api_test.go b/internal/helpers/connect_gemini_api_test.go index 5dd74af2..0d4eb40c 100644 --- a/internal/helpers/connect_gemini_api_test.go +++ b/internal/helpers/connect_gemini_api_test.go @@ -15,9 +15,12 @@ package helpers import ( "context" + "encoding/base64" "encoding/json" "net/http" "net/http/httptest" + "os" + "path/filepath" "strings" "testing" ) @@ -44,6 +47,75 @@ func TestGeminiChannelUsesAPIWithoutLocalCLI(t *testing.T) { } } +func TestGeminiAPIForwarderSendsSmallAttachmentInline(t *testing.T) { + clearChannelEnv(t) + path := filepath.Join(t.TempDir(), "voice.mp3") + wantBytes := []byte("real-audio-bytes") + if err := os.WriteFile(path, wantBytes, 0o600); err != nil { + t.Fatal(err) + } + var got geminiPart + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req geminiGenerateContentRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Errorf("decode request: %v", err) + } + if len(req.Contents) == 1 && len(req.Contents[0].Parts) == 2 { + got = req.Contents[0].Parts[1] + } + _, _ = w.Write([]byte(`{"candidates":[{"content":{"parts":[{"text":"ok"}]}}]}`)) + })) + defer ts.Close() + f := &geminiAPIForwarder{model: "gemini-test", apiKey: "key", baseURL: ts.URL, httpClient: ts.Client()} + _, err := f.forwardWithAttachments(context.Background(), "conv", "转写", []connectMediaAttachment{{LocalPath: path, FileName: "voice.mp3", MediaType: "audio"}}) + if err != nil { + t.Fatal(err) + } + if got.InlineData == nil || got.InlineData.MIMEType != "audio/mpeg" { + t.Fatalf("inlineData = %#v", got.InlineData) + } + decoded, err := base64.StdEncoding.DecodeString(got.InlineData.Data) + if err != nil || string(decoded) != string(wantBytes) { + t.Fatalf("inline bytes = %q, err=%v", decoded, err) + } +} + +func TestGeminiAPIForwarderUploadsLargeAttachment(t *testing.T) { + clearChannelEnv(t) + path := filepath.Join(t.TempDir(), "video.mp4") + if err := os.WriteFile(path, make([]byte, geminiInlineRawLimit+1), 0o600); err != nil { + t.Fatal(err) + } + var started, uploaded, generated bool + var ts *httptest.Server + ts = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/upload/v1beta/files": + started = r.Header.Get("X-Goog-Upload-Command") == "start" + w.Header().Set("X-Goog-Upload-URL", ts.URL+"/upload-session") + case "/upload-session": + uploaded = r.Header.Get("X-Goog-Upload-Command") == "upload, finalize" + _, _ = w.Write([]byte(`{"file":{"name":"files/1","uri":"https://files.example/1","mimeType":"video/mp4","state":"ACTIVE"}}`)) + case "/v1beta/models/gemini-test:generateContent": + var req geminiGenerateContentRequest + _ = json.NewDecoder(r.Body).Decode(&req) + generated = len(req.Contents) == 1 && len(req.Contents[0].Parts) == 2 && req.Contents[0].Parts[1].FileData != nil && req.Contents[0].Parts[1].FileData.FileURI == "https://files.example/1" + _, _ = w.Write([]byte(`{"candidates":[{"content":{"parts":[{"text":"ok"}]}}]}`)) + default: + http.NotFound(w, r) + } + })) + defer ts.Close() + f := &geminiAPIForwarder{model: "gemini-test", apiKey: "key", baseURL: ts.URL + "/v1beta", httpClient: ts.Client()} + _, err := f.forwardWithAttachments(context.Background(), "conv", "分析", []connectMediaAttachment{{LocalPath: path, FileName: "video.mp4", MediaType: "video"}}) + if err != nil { + t.Fatal(err) + } + if !started || !uploaded || !generated { + t.Fatalf("started=%v uploaded=%v generated=%v", started, uploaded, generated) + } +} + func TestGeminiAPIForwarderForward(t *testing.T) { clearChannelEnv(t) var gotPath, gotKey, gotText string diff --git a/internal/helpers/connect_media.go b/internal/helpers/connect_media.go index 340a9794..b9e94e80 100644 --- a/internal/helpers/connect_media.go +++ b/internal/helpers/connect_media.go @@ -18,19 +18,52 @@ import ( "encoding/json" "fmt" "io" + "mime" "net/http" "net/url" "os" "path/filepath" "strconv" "strings" + "time" "github.com/google/uuid" ) -// mediaMaxDownloadBytes caps a single inbound image download (screenshots are -// well under this; the cap is a hostile-input guard). -const mediaMaxDownloadBytes = 20 << 20 +// mediaMaxDownloadBytes caps a single inbound attachment download. Real merged +// forwards can contain high-resolution screen recordings above 250 MiB, so the +// connector allows up to 512 MiB but rejects larger payloads explicitly. It +// must never silently truncate a file and then tell an agent it has the +// original content. +const mediaMaxDownloadBytes = 512 << 20 + +const mediaDownloadTimeout = 5 * time.Minute + +// connectAttachmentMIME returns the best MIME type available for a downloaded +// attachment. DingTalk's download API returns an opaque URL, so the connector +// must recover the type from the preserved filename and, when necessary, the +// actual bytes before handing it to a multimodal backend. +func connectAttachmentMIME(path string) string { + if typ := mime.TypeByExtension(strings.ToLower(filepath.Ext(path))); typ != "" { + typ = strings.TrimSpace(strings.SplitN(typ, ";", 2)[0]) + // Generic .bin paths are common for DingTalk voice messages. Sniff their + // bytes instead of telling a multimodal backend they are opaque binary. + if typ != "application/octet-stream" { + return typ + } + } + f, err := os.Open(path) + if err != nil { + return "application/octet-stream" + } + defer f.Close() + buf := make([]byte, 512) + n, _ := f.Read(buf) + if n == 0 { + return "application/octet-stream" + } + return http.DetectContentType(buf[:n]) +} // pictureDownloadCode digs the downloadCode out of a picture callback's // loosely-typed content payload (the stream SDK models Content as @@ -48,6 +81,26 @@ func pictureDownloadCode(content interface{}) string { return "" } +func stringField(m map[string]interface{}, keys ...string) string { + for _, key := range keys { + if value, ok := m[key].(string); ok && strings.TrimSpace(value) != "" { + return strings.TrimSpace(value) + } + } + return "" +} + +// explicitPictureDownloadCode only accepts the picture-specific field. A +// generic downloadCode can represent any kind of attachment and must not be +// classified as an image merely because it is downloadable. +func explicitPictureDownloadCode(content interface{}) string { + m, ok := content.(map[string]interface{}) + if !ok { + return "" + } + return stringField(m, "pictureDownloadCode") +} + // richTextPictureDownloadCodes returns the media download codes embedded in a // msgtype="richText" callback, preserving the node order. DingTalk represents // an inline picture as a richText node instead of a top-level picture message: @@ -57,33 +110,55 @@ func pictureDownloadCode(content interface{}) string { // The two code fields identify the same picture; pictureDownloadCode handles // their precedence and returns only one code per node. func richTextPictureDownloadCodes(content interface{}) []string { + pictures, _ := richTextInboundMedia(content) + return pictures +} + +// richTextInboundMedia applies the same capability-based discovery to every +// inline node. A non-picture node with a locator is preserved as an +// attachment instead of being discarded because its type is new or unknown. +func richTextInboundMedia(content interface{}) (pictureCodes []string, files []fileInboundInfo) { m, ok := content.(map[string]interface{}) if !ok { - return nil + return nil, nil } items, ok := m["richText"].([]interface{}) if !ok { - return nil + return nil, nil } - codes := make([]string, 0) + seenPictures := make(map[string]struct{}) + seenFiles := make(map[string]struct{}) for _, item := range items { node, ok := item.(map[string]interface{}) if !ok { continue } - code := pictureDownloadCode(node) - if code == "" { + typeHint := strings.ToLower(stringField(node, "type", "msgType", "msgtype")) + code := "" + if explicitPictureDownloadCode(node) != "" { + code = pictureDownloadCode(node) + } + if code == "" && (typeHint == "" || typeHint == "picture" || typeHint == "image") { + code = pictureDownloadCode(node) + } + if code != "" { + if _, exists := seenPictures[code]; !exists { + seenPictures[code] = struct{}{} + pictureCodes = append(pictureCodes, code) + } continue } - // A richText node with a picture download code is actionable even if an - // older callback omitted type. When type is present, reject unrelated - // node kinds so future payload fields are not mistaken for pictures. - if typ, exists := node["type"].(string); exists && strings.TrimSpace(typ) != "" && !strings.EqualFold(strings.TrimSpace(typ), "picture") { + info := parseTypedFileInbound(typeHint, node) + if !info.hasActionable() { continue } - codes = append(codes, code) + key := fileInboundKey(info) + if _, exists := seenFiles[key]; !exists { + seenFiles[key] = struct{}{} + files = append(files, info) + } } - return codes + return pictureCodes, files } // fileInboundInfo carries everything a msgtype="file" callback might expose. @@ -94,16 +169,29 @@ func richTextPictureDownloadCodes(content interface{}) []string { // recognisable or the connector silently drops legitimate file messages. type fileInboundInfo struct { DownloadCode string - FileName string - FileType string - FilePath string - DentryID int64 - SpaceID int64 - FileSize int64 + // MediaID + message/conversation IDs identify media embedded in a + // forwarded chat record. DingTalk's Stream callback can erase these into + // unknownMsgType, while the user-state message API still preserves them. + MediaID string + OpenMessageID string + OpenConversationID string + // FileID is the dentryUuid returned by the user-state conversation API for + // forwarded files. It is resolved through drive.download_file. + FileID string + FileName string + FileType string + FilePath string + MediaType string + DentryID int64 + SpaceID int64 + FileSize int64 } func (f fileInboundInfo) hasActionable() bool { - return strings.TrimSpace(f.DownloadCode) != "" || (f.DentryID != 0 && f.SpaceID != 0) + return strings.TrimSpace(f.DownloadCode) != "" || + (strings.TrimSpace(f.MediaID) != "" && strings.TrimSpace(f.OpenMessageID) != "" && strings.TrimSpace(f.OpenConversationID) != "") || + strings.TrimSpace(f.FileID) != "" || + (f.DentryID != 0 && f.SpaceID != 0) } // parseFileInbound reads every relevant field out of a file callback's @@ -117,21 +205,15 @@ func parseFileInbound(content interface{}) fileInboundInfo { if !ok { return info } - for _, key := range []string{"downloadCode", "fileDownloadCode"} { - if v, ok := m[key].(string); ok && strings.TrimSpace(v) != "" { - info.DownloadCode = strings.TrimSpace(v) - break - } - } - if v, ok := m["fileName"].(string); ok { - info.FileName = strings.TrimSpace(v) - } - if v, ok := m["fileType"].(string); ok { - info.FileType = strings.TrimSpace(v) - } - if v, ok := m["filePath"].(string); ok { - info.FilePath = strings.TrimSpace(v) - } + info.DownloadCode = stringField(m, "downloadCode", "fileDownloadCode") + info.MediaID = stringField(m, "mediaId", "mediaID") + info.OpenMessageID = stringField(m, "openMessageId", "openMessageID") + info.OpenConversationID = stringField(m, "openConversationId", "openConversationID") + info.FileID = stringField(m, "fileId", "fileID", "dentryUuid", "dentryUUID") + info.FileName = stringField(m, "fileName", "name") + info.FileType = stringField(m, "fileType") + info.FilePath = stringField(m, "filePath") + info.MediaType = stringField(m, "mediaType") info.DentryID = readInt64Field(m, "dentryId", "dentryID") info.SpaceID = readInt64Field(m, "spaceId", "spaceID") info.FileSize = readInt64Field(m, "fileSize", "size") @@ -141,6 +223,213 @@ func parseFileInbound(content interface{}) fileInboundInfo { return info } +// inboundMediaType normalizes the callback spellings used for downloadable +// non-picture media. The returned value is only used to make the agent prompt +// precise; download authorization still comes exclusively from downloadCode +// or dentryId+spaceId. +func inboundMediaType(msgtype string) string { + switch strings.ToLower(strings.TrimSpace(msgtype)) { + case "image", "picture": + return "image" + case "audio", "voice": + return "audio" + case "video": + return "video" + default: + return "file" + } +} + +func parseTypedFileInbound(msgtype string, content interface{}) fileInboundInfo { + info := parseFileInbound(content) + if mediaType := strings.TrimSpace(msgtype); mediaType != "" { + info.MediaType = inboundMediaType(mediaType) + } + if info.MediaType == "" || info.MediaType == "file" { + info.MediaType = mediaTypeFromFileName(info.FileName) + } + if info.FileName == "未知文件" { + switch info.MediaType { + case "audio": + info.FileName = "语音消息" + case "video": + info.FileName = "视频消息" + } + } + return info +} + +func fileInboundKey(info fileInboundInfo) string { + switch { + case strings.TrimSpace(info.DownloadCode) != "": + return "download:" + strings.TrimSpace(info.DownloadCode) + case strings.TrimSpace(info.MediaID) != "": + return "media:" + strings.TrimSpace(info.MediaID) + ":" + strings.TrimSpace(info.OpenMessageID) + ":" + strings.TrimSpace(info.OpenConversationID) + case strings.TrimSpace(info.FileID) != "": + return "file:" + strings.TrimSpace(info.FileID) + case info.DentryID != 0 && info.SpaceID != 0: + return fmt.Sprintf("dentry:%d:%d", info.SpaceID, info.DentryID) + default: + return "" + } +} + +// chatRecordInboundMedia extracts every actionable attachment that DingTalk +// preserved in a msgtype=chatRecord callback. The observed callback encodes +// the record array as a JSON string under content.chatRecord; accepting an +// already-decoded array as well keeps the parser compatible with SDK changes. +// +// Some forwarded entries arrive as {"msgType":"unknownMsgType"} with no +// message id, download code, or storage id. Those entries are counted for +// diagnostics but cannot be recovered by the connector because the callback +// contains no locator for the original bytes. +func chatRecordInboundMedia(content interface{}) (pictureCodes []string, files []fileInboundInfo, unrecoverableCount int) { + m, ok := content.(map[string]interface{}) + if !ok { + return nil, nil, 0 + } + + var entries []interface{} + switch raw := m["chatRecord"].(type) { + case string: + if err := json.Unmarshal([]byte(raw), &entries); err != nil { + return nil, nil, 0 + } + case []interface{}: + entries = raw + default: + // A few callback variants call the decoded array "contents". + if decoded, ok := m["contents"].([]interface{}); ok { + entries = decoded + } + } + + seenPictures := make(map[string]struct{}) + seenFiles := make(map[string]struct{}) + for _, entry := range entries { + node, ok := entry.(map[string]interface{}) + if !ok { + continue + } + msgtype := "" + for _, key := range []string{"msgType", "msgtype", "type"} { + if value, ok := node[key].(string); ok && strings.TrimSpace(value) != "" { + msgtype = strings.ToLower(strings.TrimSpace(value)) + break + } + } + // Attachment discovery is capability-based. The nested msgType is a + // classification hint only; a future/unknown type with a valid locator + // must still reach the backend with its original bytes. + pictureCode := "" + if explicitPictureDownloadCode(node) != "" { + pictureCode = pictureDownloadCode(node) + } + if pictureCode == "" && (msgtype == "picture" || msgtype == "image") { + pictureCode = pictureDownloadCode(node) + } + if pictureCode != "" { + if _, exists := seenPictures[pictureCode]; !exists { + seenPictures[pictureCode] = struct{}{} + pictureCodes = append(pictureCodes, pictureCode) + } + continue + } + + info := parseTypedFileInbound(msgtype, node) + if info.hasActionable() { + key := fileInboundKey(info) + if _, exists := seenFiles[key]; !exists { + seenFiles[key] = struct{}{} + files = append(files, info) + } + continue + } + if msgtype == "unknownmsgtype" { + unrecoverableCount++ + } + } + return pictureCodes, files, unrecoverableCount +} + +func hasChatRecordPayload(content interface{}) bool { + m, ok := content.(map[string]interface{}) + if !ok { + return false + } + switch raw := m["chatRecord"].(type) { + case string: + return strings.TrimSpace(raw) != "" + case []interface{}: + return true + } + _, hasDecodedContents := m["contents"].([]interface{}) + return hasDecodedContents +} + +// callbackInboundMedia discovers downloadable payloads from their locator +// fields instead of an allowlist of msgtype values. msgtype is retained only +// as a media classification hint, so newly introduced message types are +// forwarded immediately without requiring a connector release. +func callbackInboundMedia(msgtype string, content interface{}) (pictureCodes []string, files []fileInboundInfo, unrecoverableCount int) { + seenPictures := make(map[string]struct{}) + seenFiles := make(map[string]struct{}) + addPicture := func(code string) { + code = strings.TrimSpace(code) + if code == "" { + return + } + if _, exists := seenPictures[code]; exists { + return + } + seenPictures[code] = struct{}{} + pictureCodes = append(pictureCodes, code) + } + addFile := func(info fileInboundInfo) { + if !info.hasActionable() { + return + } + key := fileInboundKey(info) + if key == "" { + return + } + if _, picture := seenPictures[info.DownloadCode]; picture { + return + } + if _, exists := seenFiles[key]; exists { + return + } + seenFiles[key] = struct{}{} + files = append(files, info) + } + + richPictures, richFiles := richTextInboundMedia(content) + for _, code := range richPictures { + addPicture(code) + } + for _, info := range richFiles { + addFile(info) + } + pictureCode := "" + if explicitPictureDownloadCode(content) != "" { + pictureCode = pictureDownloadCode(content) + } + if pictureCode == "" && (strings.EqualFold(msgtype, "picture") || strings.EqualFold(msgtype, "image")) { + pictureCode = pictureDownloadCode(content) + } + addPicture(pictureCode) + addFile(parseTypedFileInbound(msgtype, content)) + + nestedPictures, nestedFiles, nestedUnknown := chatRecordInboundMedia(content) + for _, code := range nestedPictures { + addPicture(code) + } + for _, info := range nestedFiles { + addFile(info) + } + return pictureCodes, files, nestedUnknown +} + // readInt64Field pulls an int64 out of the loose content map under any of the // provided keys, tolerating JSON string / float64 / int64 / json.Number. func readInt64Field(m map[string]interface{}, keys ...string) int64 { @@ -196,6 +485,19 @@ func summarizeContent(content interface{}) string { return s } +// rawCallbackPrompt preserves message types whose payload is meaningful but +// has no locally recognised text/media shape (for example msgtype=chatRecord). +// The connector should not decide that such messages are empty: forwarding +// the type and JSON payload lets the backend model interpret new and complex +// DingTalk message formats without waiting for a CLI-side parser update. +func rawCallbackPrompt(msgtype string, content interface{}) string { + b, err := json.Marshal(content) + if err != nil { + b = []byte(fmt.Sprintf("%v", content)) + } + return fmt.Sprintf("用户发送了一条钉钉消息，msgtype=%q。请解析以下原始消息 JSON，提取有用信息并处理用户意图：\n%s", strings.TrimSpace(msgtype), b) +} + // extractCallbackText pulls the visible text out of a structured-text callback // payload (msgtype=richText / markdown / etc.) for the case where the SDK's // data.Text.Content is empty. This matters because `dws chat message send @@ -210,7 +512,7 @@ func extractCallbackText(content interface{}) string { case map[string]interface{}: // Common shapes: {"text":"..."}, {"title":"...","text":"..."}, // {"content":"..."}, richText {"richText":[{"text":"..."}]}. - for _, key := range []string{"text", "content", "markdown", "title"} { + for _, key := range []string{"text", "content", "markdown", "title", "recognition"} { if s, ok := v[key].(string); ok && strings.TrimSpace(s) != "" { return strings.TrimSpace(s) } @@ -297,6 +599,13 @@ func extractInteractiveCardText(content interface{}) string { // questions are the top Q&A inbound; without this the connector silently // drops every picture message. func (c *aiCardClient) downloadMessageFile(ctx context.Context, robotCode, downloadCode string) (string, error) { + return c.downloadMessageFileNamed(ctx, robotCode, downloadCode, "") +} + +// downloadMessageFileNamed is downloadMessageFile with an optional original +// file name. Keeping its extension materially improves audio/video/file +// handling across local agents whose tool selection depends on the path. +func (c *aiCardClient) downloadMessageFileNamed(ctx context.Context, robotCode, downloadCode, fileName string) (string, error) { raw, err := c.callRaw(ctx, http.MethodPost, "/v1.0/robot/messageFiles/download", map[string]any{ "robotCode": robotCode, "downloadCode": downloadCode, @@ -314,7 +623,7 @@ func (c *aiCardClient) downloadMessageFile(ctx context.Context, robotCode, downl if err != nil { return "", err } - resp, err := c.httpClient.Do(req) + resp, err := connectMediaDownloadClient(c.httpClient).Do(req) if err != nil { return "", err } @@ -322,18 +631,19 @@ func (c *aiCardClient) downloadMessageFile(ctx context.Context, robotCode, downl if resp.StatusCode >= 400 { return "", fmt.Errorf("媒体下载 HTTP %d", resp.StatusCode) } + if resp.ContentLength > mediaMaxDownloadBytes { + return "", fmt.Errorf("媒体文件过大：%d 字节，最大允许 %d 字节", resp.ContentLength, mediaMaxDownloadBytes) + } dir := filepath.Join(os.TempDir(), "dws-connect-media") if err := os.MkdirAll(dir, 0o755); err != nil { return "", err } - dest := filepath.Join(dir, uuid.NewString()+mediaExt(parsed.DownloadUrl, resp.Header.Get("Content-Type"))) - f, err := os.Create(dest) - if err != nil { - return "", err + ext := filepath.Ext(strings.TrimSpace(fileName)) + if ext == "" { + ext = mediaExt(parsed.DownloadUrl, resp.Header.Get("Content-Type")) } - defer f.Close() - if _, err := io.Copy(f, io.LimitReader(resp.Body, mediaMaxDownloadBytes)); err != nil { - _ = os.Remove(dest) + dest := filepath.Join(dir, uuid.NewString()+ext) + if err := writeCompleteMediaFile(dest, resp.Body); err != nil { return "", err } return dest, nil @@ -382,7 +692,7 @@ func (c *aiCardClient) downloadDentryFile(ctx context.Context, spaceID, dentryID for k, v := range parsed.HeadersMap { req.Header.Set(k, v) } - resp, err := c.httpClient.Do(req) + resp, err := connectMediaDownloadClient(c.httpClient).Do(req) if err != nil { return "", err } @@ -390,6 +700,9 @@ func (c *aiCardClient) downloadDentryFile(ctx context.Context, spaceID, dentryID if resp.StatusCode >= 400 { return "", fmt.Errorf("钉盘文件下载 HTTP %d", resp.StatusCode) } + if resp.ContentLength > mediaMaxDownloadBytes { + return "", fmt.Errorf("钉盘文件过大：%d 字节，最大允许 %d 字节", resp.ContentLength, mediaMaxDownloadBytes) + } dir := filepath.Join(os.TempDir(), "dws-connect-media") if err := os.MkdirAll(dir, 0o755); err != nil { @@ -400,18 +713,60 @@ func (c *aiCardClient) downloadDentryFile(ctx context.Context, spaceID, dentryID ext = mediaExt(parsed.ResourceURL, resp.Header.Get("Content-Type")) } dest := filepath.Join(dir, uuid.NewString()+ext) - f, err := os.Create(dest) - if err != nil { - return "", err - } - defer f.Close() - if _, err := io.Copy(f, io.LimitReader(resp.Body, mediaMaxDownloadBytes)); err != nil { - _ = os.Remove(dest) + if err := writeCompleteMediaFile(dest, resp.Body); err != nil { return "", err } return dest, nil } +// writeCompleteMediaFile writes at most mediaMaxDownloadBytes and verifies the +// stream ended. Reading one byte past the cap distinguishes an exact-size file +// from a larger file; the latter is removed instead of leaving a corrupt local +// artifact that an agent could mistake for the original. +func writeCompleteMediaFile(dest string, src io.Reader) error { + f, err := os.Create(dest) + if err != nil { + return err + } + n, copyErr := io.Copy(f, io.LimitReader(src, mediaMaxDownloadBytes+1)) + closeErr := f.Close() + if copyErr != nil { + _ = os.Remove(dest) + return copyErr + } + if closeErr != nil { + _ = os.Remove(dest) + return closeErr + } + if n > mediaMaxDownloadBytes { + _ = os.Remove(dest) + return fmt.Errorf("媒体文件超过最大允许大小 %d 字节，未保存截断文件", mediaMaxDownloadBytes) + } + return nil +} + +func connectMediaDownloadClient(base *http.Client) *http.Client { + if base == nil { + return &http.Client{Timeout: mediaDownloadTimeout} + } + clone := *base + if clone.Timeout <= 0 || clone.Timeout < mediaDownloadTimeout { + clone.Timeout = mediaDownloadTimeout + } + return &clone +} + +func cleanupConnectMediaAttachments(attachments []connectMediaAttachment) { + root := filepath.Join(os.TempDir(), "dws-connect-media") + for _, attachment := range attachments { + path := filepath.Clean(strings.TrimSpace(attachment.LocalPath)) + if path == "." || filepath.Dir(path) != root { + continue + } + _ = os.Remove(path) + } +} + // mediaExt picks a file extension from the response content type, falling // back to the URL path, then ".png" (DingTalk screenshots default to png). func mediaExt(rawURL, contentType string) string { diff --git a/internal/helpers/connect_media_test.go b/internal/helpers/connect_media_test.go index d682fd13..d8238ca8 100644 --- a/internal/helpers/connect_media_test.go +++ b/internal/helpers/connect_media_test.go @@ -16,9 +16,11 @@ package helpers import ( "context" "encoding/json" + "fmt" "net/http" "net/http/httptest" "os" + "path/filepath" "strings" "testing" ) @@ -80,6 +82,18 @@ func TestRichTextPictureDownloadCodesUnknownShape(t *testing.T) { } } +func TestCallbackInboundMediaPreservesFutureRichTextAttachment(t *testing.T) { + pictures, files, unknown := callbackInboundMedia("renamedRichEnvelope", map[string]interface{}{ + "richText": []interface{}{ + map[string]interface{}{"type": "text", "text": "附件如下"}, + map[string]interface{}{"type": "futureInlineBinary", "downloadCode": "inline-1", "fileName": "demo.mp4"}, + }, + }) + if len(pictures) != 0 || len(files) != 1 || files[0].DownloadCode != "inline-1" || files[0].MediaType != "video" || unknown != 0 { + t.Fatalf("pictures=%v files=%#v unknown=%d", pictures, files, unknown) + } +} + // TestExtractCallbackText covers the markdown / richText fallback path used // when SDK data.Text.Content is empty. This is the recovery path for // `dws chat message send --group ... --text ...` (defaults to msgType=markdown) @@ -122,6 +136,156 @@ func TestExtractCallbackText(t *testing.T) { } } +func TestRawCallbackPromptPreservesChatRecordPayload(t *testing.T) { + content := map[string]interface{}{ + "title": "转发的聊天记录", + "contents": []interface{}{ + map[string]interface{}{"senderName": "张三", "text": "请汇总本周风险"}, + map[string]interface{}{"senderName": "李四", "text": "发布窗口需要延期"}, + }, + } + + got := rawCallbackPrompt(" chatRecord ", content) + for _, want := range []string{ + `msgtype="chatRecord"`, + `"senderName":"张三"`, + `"text":"请汇总本周风险"`, + `"text":"发布窗口需要延期"`, + } { + if !strings.Contains(got, want) { + t.Fatalf("rawCallbackPrompt() missing %q:\n%s", want, got) + } + } +} + +func TestRawCallbackPromptForUnknownEmptyPayload(t *testing.T) { + got := rawCallbackPrompt("futureMessageType", nil) + if !strings.Contains(got, `msgtype="futureMessageType"`) || !strings.HasSuffix(got, "\nnull") { + t.Fatalf("rawCallbackPrompt() = %q, want message type and null JSON payload", got) + } +} + +func TestChatRecordInboundMediaExtractsEveryRecoverableAttachment(t *testing.T) { + record := []interface{}{ + map[string]interface{}{"msgType": "picture", "downloadCode": "pic-1"}, + map[string]interface{}{"msgType": "picture", "downloadCode": "pic-1"}, // duplicate + map[string]interface{}{"msgType": "audio", "downloadCode": "audio-1", "recognition": "语音转写"}, + map[string]interface{}{"msgType": "video", "downloadCode": "video-1", "fileName": "demo.mov"}, + map[string]interface{}{"msgType": "file", "downloadCode": "file-1", "fileName": "report.md"}, + map[string]interface{}{"msgType": "file", "dentryId": "123", "spaceId": "456", "fileName": "spec.pdf"}, + map[string]interface{}{"msgType": "unknownMsgType"}, + map[string]interface{}{"msgType": "unknownMsgType"}, + map[string]interface{}{"msgType": "text", "content": "请分析这些附件"}, + } + raw, err := json.Marshal(record) + if err != nil { + t.Fatal(err) + } + + pictures, files, unknown := chatRecordInboundMedia(map[string]interface{}{"chatRecord": string(raw)}) + if len(pictures) != 1 || pictures[0] != "pic-1" { + t.Fatalf("pictures = %v, want [pic-1]", pictures) + } + if unknown != 2 { + t.Fatalf("unknownCount = %d, want 2", unknown) + } + if len(files) != 4 { + t.Fatalf("files = %#v, want 4 attachments", files) + } + wants := []struct { + mediaType string + code string + name string + }{ + {"audio", "audio-1", "语音消息"}, + {"video", "video-1", "demo.mov"}, + {"file", "file-1", "report.md"}, + {"file", "", "spec.pdf"}, + } + for i, want := range wants { + if files[i].MediaType != want.mediaType || files[i].DownloadCode != want.code || files[i].FileName != want.name { + t.Fatalf("files[%d] = %#v, want type=%q code=%q name=%q", i, files[i], want.mediaType, want.code, want.name) + } + } + if files[3].DentryID != 123 || files[3].SpaceID != 456 { + t.Fatalf("dentry attachment = %#v, want dentry=123 space=456", files[3]) + } +} + +func TestChatRecordInboundMediaMatchesObservedDegradedCallback(t *testing.T) { + record := `[{"msgType":"picture","downloadCode":"pic-live"},{"msgType":"unknownMsgType"},{"msgType":"unknownMsgType"},{"msgType":"text","content":"[合并的聊天记录]"},{"msgType":"unknownMsgType"}]` + pictures, files, unknown := chatRecordInboundMedia(map[string]interface{}{"chatRecord": record}) + if len(pictures) != 1 || pictures[0] != "pic-live" || len(files) != 0 || unknown != 3 { + t.Fatalf("pictures=%v files=%v unknown=%d, want one picture, no recoverable files, three unknowns", pictures, files, unknown) + } +} + +func TestChatRecordInboundMediaRecoversUnknownTypeWhenLocatorSurvives(t *testing.T) { + record := `[{"msgType":"unknownMsgType","downloadCode":"opaque-1","fileName":"payload.bin"}]` + pictures, files, unknown := chatRecordInboundMedia(map[string]interface{}{"chatRecord": record}) + if len(pictures) != 0 || len(files) != 1 || files[0].DownloadCode != "opaque-1" || files[0].FileName != "payload.bin" || unknown != 0 { + t.Fatalf("pictures=%v files=%#v unknown=%d", pictures, files, unknown) + } +} + +func TestChatRecordInboundMediaPreservesFutureTypeWithLocator(t *testing.T) { + record := `[{"msgType":"futureBinaryEnvelope","downloadCode":"future-1","fileName":"clip.webm"}]` + pictures, files, unknown := chatRecordInboundMedia(map[string]interface{}{"chatRecord": record}) + if len(pictures) != 0 || len(files) != 1 || files[0].DownloadCode != "future-1" || files[0].MediaType != "video" || unknown != 0 { + t.Fatalf("pictures=%v files=%#v unknown=%d", pictures, files, unknown) + } +} + +func TestCallbackInboundMediaDoesNotGateOnOuterMessageType(t *testing.T) { + pictures, files, unknown := callbackInboundMedia("futureAttachmentV2", map[string]interface{}{ + "downloadCode": "future-2", + "fileName": "voice.ogg", + }) + if len(pictures) != 0 || len(files) != 1 || files[0].DownloadCode != "future-2" || files[0].MediaType != "audio" || unknown != 0 { + t.Fatalf("pictures=%v files=%#v unknown=%d", pictures, files, unknown) + } +} + +func TestCallbackInboundMediaFindsChatRecordByShape(t *testing.T) { + content := map[string]interface{}{ + "chatRecord": `[{"msgType":"picture","downloadCode":"nested-picture"},{"msgType":"futureFile","downloadCode":"nested-file","fileName":"notes.md"},{"msgType":"unknownMsgType"}]`, + } + pictures, files, unknown := callbackInboundMedia("renamedForwardEnvelope", content) + if len(pictures) != 1 || pictures[0] != "nested-picture" { + t.Fatalf("pictures=%v, want [nested-picture]", pictures) + } + if len(files) != 1 || files[0].DownloadCode != "nested-file" || files[0].FileName != "notes.md" { + t.Fatalf("files=%#v, want nested-file", files) + } + if unknown != 1 { + t.Fatalf("unknown=%d, want 1", unknown) + } +} + +func TestHasChatRecordPayloadUsesShapeNotMessageType(t *testing.T) { + if !hasChatRecordPayload(map[string]interface{}{ + "title": "转发记录", + "chatRecord": `[{"msgType":"text","content":"hello"}]`, + }) { + t.Fatal("chatRecord JSON string should be detected by payload shape") + } + if hasChatRecordPayload(map[string]interface{}{"title": "普通卡片"}) { + t.Fatal("ordinary title payload must not be detected as a chat record") + } +} + +func TestChatRecordInboundMediaAcceptsDecodedContents(t *testing.T) { + pictures, files, unknown := chatRecordInboundMedia(map[string]interface{}{ + "contents": []interface{}{ + map[string]interface{}{"type": "image", "pictureDownloadCode": "pic-2"}, + map[string]interface{}{"type": "voice", "downloadCode": "voice-2"}, + }, + }) + if len(pictures) != 1 || pictures[0] != "pic-2" || len(files) != 1 || files[0].MediaType != "audio" || unknown != 0 { + t.Fatalf("pictures=%v files=%#v unknown=%d", pictures, files, unknown) + } +} + // TestExtractInteractiveCardText covers the bot→bot @ card: the leading // mention leaf (whose display name may contain spaces) is dropped by leaf // boundary, leaving the clean instruction. @@ -199,6 +363,61 @@ func TestDownloadMessageFile(t *testing.T) { } } +func TestDownloadMessageFileNamedPreservesOriginalExtension(t *testing.T) { + mux := http.NewServeMux() + var srv *httptest.Server + mux.HandleFunc("/v1.0/oauth2/accessToken", func(w http.ResponseWriter, r *http.Request) { + _ = json.NewEncoder(w).Encode(map[string]any{"accessToken": "tok-1", "expireIn": 7200}) + }) + mux.HandleFunc("/v1.0/robot/messageFiles/download", func(w http.ResponseWriter, r *http.Request) { + _ = json.NewEncoder(w).Encode(map[string]any{"downloadUrl": srv.URL + "/opaque.file"}) + }) + mux.HandleFunc("/opaque.file", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/octet-stream") + _, _ = w.Write([]byte("MOVDATA")) + }) + srv = httptest.NewServer(mux) + defer srv.Close() + withCardAPIBase(t, srv.URL) + + c := newAICardClient("ding-client", "ding-secret", "") + path, err := c.downloadMessageFileNamed(context.Background(), "ding-client", "video-code", "screen.mov") + if err != nil { + t.Fatalf("downloadMessageFileNamed: %v", err) + } + defer os.Remove(path) + if !strings.HasSuffix(path, ".mov") { + t.Fatalf("path = %q, want original .mov extension", path) + } + raw, err := os.ReadFile(path) + if err != nil || string(raw) != "MOVDATA" { + t.Fatalf("saved file = %q, %v", raw, err) + } +} + +func TestDownloadMessageFileRejectsKnownOversizePayload(t *testing.T) { + mux := http.NewServeMux() + var srv *httptest.Server + mux.HandleFunc("/v1.0/oauth2/accessToken", func(w http.ResponseWriter, r *http.Request) { + _ = json.NewEncoder(w).Encode(map[string]any{"accessToken": "tok-1", "expireIn": 7200}) + }) + mux.HandleFunc("/v1.0/robot/messageFiles/download", func(w http.ResponseWriter, r *http.Request) { + _ = json.NewEncoder(w).Encode(map[string]any{"downloadUrl": srv.URL + "/too-large"}) + }) + mux.HandleFunc("/too-large", func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Length", fmt.Sprintf("%d", mediaMaxDownloadBytes+1)) + w.WriteHeader(http.StatusOK) + }) + srv = httptest.NewServer(mux) + defer srv.Close() + withCardAPIBase(t, srv.URL) + + c := newAICardClient("ding-client", "ding-secret", "") + if _, err := c.downloadMessageFileNamed(context.Background(), "ding-client", "large-code", "large.mov"); err == nil || !strings.Contains(err.Error(), "文件过大") { + t.Fatalf("oversize download error = %v, want explicit size rejection", err) + } +} + func TestDownloadMessageFileNoURL(t *testing.T) { mux := http.NewServeMux() mux.HandleFunc("/v1.0/oauth2/accessToken", func(w http.ResponseWriter, r *http.Request) { @@ -231,6 +450,18 @@ func TestMediaExt(t *testing.T) { } } +func TestConnectAttachmentMIMESniffsGenericVoiceFile(t *testing.T) { + path := filepath.Join(t.TempDir(), "voice.bin") + // Ogg capture pattern plus a minimal header is sufficient for + // http.DetectContentType to identify the real container. + if err := os.WriteFile(path, append([]byte("OggS\x00\x02"), make([]byte, 506)...), 0o600); err != nil { + t.Fatal(err) + } + if got := connectAttachmentMIME(path); got != "application/ogg" { + t.Fatalf("connectAttachmentMIME() = %q, want application/ogg", got) + } +} + // TestParseFileInbound covers both callback shapes so the regression that // dropped every API-sent file (dentryId/spaceId, no downloadCode) can't // silently return: (a) client-sent shape carrying downloadCode + fileName, @@ -260,6 +491,17 @@ func TestParseFileInbound(t *testing.T) { wantName: "log.txt", wantActionable: true, }, + { + name: "media locator fields", + content: map[string]interface{}{ + "mediaId": "media-1", + "openMessageId": "message-1", + "openConversationId": "conversation-1", + "fileName": "recording.m4a", + }, + wantName: "recording.m4a", + wantActionable: true, + }, { name: "API-sent dentryId/spaceId as numbers", content: map[string]interface{}{ diff --git a/internal/helpers/connect_opencode.go b/internal/helpers/connect_opencode.go index 6485ed08..45927a89 100644 --- a/internal/helpers/connect_opencode.go +++ b/internal/helpers/connect_opencode.go @@ -24,6 +24,7 @@ import ( "io" "net" "net/http" + "net/url" "os" "os/exec" "path/filepath" @@ -99,7 +100,15 @@ func (f *opencodeForwarder) forward(ctx context.Context, convID, text string) (s return f.forwardStream(ctx, convID, text, nil) } +func (f *opencodeForwarder) forwardWithAttachments(ctx context.Context, convID, text string, attachments []connectMediaAttachment) (string, error) { + return f.forwardStreamWithAttachments(ctx, convID, text, attachments, nil) +} + func (f *opencodeForwarder) forwardStream(ctx context.Context, convID, text string, _ func(string)) (string, error) { + return f.forwardStreamWithAttachments(ctx, convID, text, nil, nil) +} + +func (f *opencodeForwarder) forwardStreamWithAttachments(ctx context.Context, convID, text string, attachments []connectMediaAttachment, _ func(string)) (string, error) { ctx, cancel := applyTimeout(ctx, f.timeout) defer cancel() @@ -107,15 +116,15 @@ func (f *opencodeForwarder) forwardStream(ctx context.Context, convID, text stri if err != nil { return "", err } - reply, err := f.forwardWithClient(ctx, client, convID, text) + reply, err := f.forwardWithClient(ctx, client, convID, text, attachments) if errors.Is(err, errOpencodeSessionMissing) && f.sessions != nil { f.sessions.reset(convID) - reply, err = f.forwardWithClient(ctx, client, convID, text) + reply, err = f.forwardWithClient(ctx, client, convID, text, attachments) } return reply, err } -func (f *opencodeForwarder) forwardWithClient(ctx context.Context, client *opencodeHTTPClient, convID, text string) (string, error) { +func (f *opencodeForwarder) forwardWithClient(ctx context.Context, client *opencodeHTTPClient, convID, text string, attachments []connectMediaAttachment) (string, error) { sessionID := "" if f.sessions != nil { sessionID = f.sessions.id(convID) @@ -130,7 +139,7 @@ func (f *opencodeForwarder) forwardWithClient(ctx context.Context, client *openc f.sessions.set(convID, sessionID) } } - reply, err := client.sendMessage(ctx, sessionID, text, f.model) + reply, err := client.sendMessageWithAttachments(ctx, sessionID, text, f.model, attachments) if err != nil { return "", err } @@ -459,9 +468,27 @@ func (c *opencodeHTTPClient) deleteSession(ctx context.Context, sessionID string return nil } -func (c *opencodeHTTPClient) sendMessage(ctx context.Context, sessionID, text, model string) (string, error) { +func (c *opencodeHTTPClient) sendMessageWithAttachments(ctx context.Context, sessionID, text, model string, attachments []connectMediaAttachment) (string, error) { + parts := []map[string]any{{"type": "text", "text": text}} + for _, attachment := range attachments { + path := strings.TrimSpace(attachment.LocalPath) + if path == "" { + continue + } + fileURL := (&url.URL{Scheme: "file", Path: path}).String() + name := strings.TrimSpace(attachment.FileName) + if name == "" { + name = filepath.Base(path) + } + parts = append(parts, map[string]any{ + "type": "file", + "url": fileURL, + "filename": name, + "mime": connectAttachmentMIME(path), + }) + } body := map[string]any{ - "parts": []map[string]any{{"type": "text", "text": text}}, + "parts": parts, } if m := opencodeModelRef(model); m != nil { body["model"] = m diff --git a/internal/helpers/connect_opencode_test.go b/internal/helpers/connect_opencode_test.go index 7714de5a..230012bc 100644 --- a/internal/helpers/connect_opencode_test.go +++ b/internal/helpers/connect_opencode_test.go @@ -19,12 +19,53 @@ import ( "fmt" "net/http" "net/http/httptest" + "os" "path/filepath" "strings" "testing" "time" ) +func TestOpencodeForwarderSendsNativeFileParts(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "evidence.png") + if err := os.WriteFile(path, []byte("png-bytes"), 0o600); err != nil { + t.Fatal(err) + } + var gotParts []map[string]any + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch { + case r.Method == http.MethodGet && r.URL.Path == "/global/health": + _, _ = w.Write([]byte(`{"healthy":true}`)) + case r.Method == http.MethodPost && r.URL.Path == "/session": + _, _ = w.Write([]byte(`{"id":"ses_media"}`)) + case r.Method == http.MethodPost && r.URL.Path == "/session/ses_media/message": + var body struct { + Parts []map[string]any `json:"parts"` + } + _ = json.NewDecoder(r.Body).Decode(&body) + gotParts = body.Parts + _, _ = w.Write([]byte(`{"parts":[{"type":"text","text":"ok"}]}`)) + default: + http.NotFound(w, r) + } + })) + defer ts.Close() + f := &opencodeForwarder{ + bin: "opencode", timeout: 5 * time.Second, workDir: dir, + sessions: newOpencodeSessions(""), + server: &opencodeServer{baseURL: ts.URL, httpClient: ts.Client()}, + } + _, err := f.forwardWithAttachments(context.Background(), "conv", "看图", []connectMediaAttachment{{LocalPath: path, FileName: "evidence.png", MediaType: "image"}}) + if err != nil { + t.Fatal(err) + } + if len(gotParts) != 2 || gotParts[1]["type"] != "file" || gotParts[1]["filename"] != "evidence.png" || !strings.HasPrefix(fmt.Sprint(gotParts[1]["url"]), "file://") { + t.Fatalf("parts = %#v", gotParts) + } +} + func TestOpencodeForwarderUsesServerSessionAPI(t *testing.T) { dir := t.TempDir() storePath := filepath.Join(dir, "opencode-sessions.json") diff --git a/internal/helpers/connect_qoder_stream.go b/internal/helpers/connect_qoder_stream.go index ff412013..d31d3689 100644 --- a/internal/helpers/connect_qoder_stream.go +++ b/internal/helpers/connect_qoder_stream.go @@ -80,6 +80,58 @@ func (f *qoderStreamForwarder) forward(ctx context.Context, convID, text string) return f.forwardStream(ctx, convID, text, nil) } +// forwardWithAttachments uses qodercli's native --attachment transport for a +// media turn. The persistent stream-json protocol has no documented file-part +// shape, so sending the same session through a one-shot CLI process is the +// only reliable way to provide the original bytes without enabling broad file +// tools for every ordinary chat message. +func (f *qoderStreamForwarder) forwardWithAttachments(ctx context.Context, convID, text string, attachments []connectMediaAttachment) (string, error) { + if len(attachments) == 0 { + return f.forward(ctx, convID, text) + } + ctx, cancel := applyTimeout(ctx, f.timeout) + defer cancel() + + f.mu.Lock() + defer f.mu.Unlock() + args := []string{"--print", "--output-format", "text", "--max-turns", "30"} + if f.sessions != nil { + args = append(args, f.sessions.args(convID)...) + } + if f.yolo { + args = append(args, "--permission-mode", "bypass_permissions", "--dangerously-skip-permissions") + } else { + args = append(args, "--system-prompt", "", "--setting-sources", "", "--tools", "") + } + if f.model != "" { + args = append(args, "--model", f.model) + } + for _, attachment := range attachments { + if path := strings.TrimSpace(attachment.LocalPath); path != "" { + args = append(args, "--attachment", path) + } + } + args = append(args, "-p", text) + cmd := exec.CommandContext(ctx, f.bin, args...) + cmd.Dir = f.cwd() + cmd.Env = append(os.Environ(), f.env...) + out, err := cmd.Output() + reply := strings.TrimSpace(string(out)) + if reply != "" && !agentReplyIsError(reply) { + return brandReply(f.name, reply), nil + } + if reply != "" { + return agentBackendErrorReply(reply), nil + } + if err != nil { + if f.sessions != nil { + f.sessions.reset(convID) + } + return "", fmt.Errorf("本地 %s agent 附件调用失败：%s", f.name, truncateRunes(execErrorMessage(err), 300)) + } + return "（本地 agent 无文本输出）", nil +} + func (f *qoderStreamForwarder) forwardStream(ctx context.Context, convID, text string, onDelta func(string)) (string, error) { ctx, cancel := applyTimeout(ctx, f.timeout) defer cancel() diff --git a/internal/helpers/connect_qoder_stream_test.go b/internal/helpers/connect_qoder_stream_test.go index 90a933bf..8ce8d507 100644 --- a/internal/helpers/connect_qoder_stream_test.go +++ b/internal/helpers/connect_qoder_stream_test.go @@ -65,11 +65,7 @@ if "--input-format" in sys.argv: }), flush=True) else: prompt = sys.argv[-1] if len(sys.argv) > 1 else "" - print(json.dumps({ - "type": "result", - "subtype": "success", - "message": {"content": [{"type": "text", "text": "one-shot " + prompt}]}, - })) + print("one-shot " + prompt) ` path := filepath.Join(dir, "qodercli") if err := os.WriteFile(path, []byte(script), 0o755); err != nil { @@ -78,6 +74,35 @@ else: return logPath } +func TestQoderForwarderUsesNativeAttachmentFlag(t *testing.T) { + t.Setenv("DWS_CONFIG_DIR", t.TempDir()) + stubDir := t.TempDir() + logPath := writeQoderStreamStub(t, stubDir) + t.Setenv("DWS_QODER_STUB_LOG", logPath) + attachmentPath := filepath.Join(t.TempDir(), "forwarded.mov") + if err := os.WriteFile(attachmentPath, []byte("video-bytes"), 0o600); err != nil { + t.Fatal(err) + } + f := &qoderStreamForwarder{ + name: "qoderwork", bin: filepath.Join(stubDir, "qodercli"), + timeout: 5 * time.Second, sessions: newConvSessions(""), + } + reply, err := f.forwardWithAttachments(context.Background(), "conv", "分析视频", []connectMediaAttachment{{LocalPath: attachmentPath, FileName: "forwarded.mov", MediaType: "video"}}) + if err != nil { + t.Fatal(err) + } + if reply != "one-shot 分析视频" { + t.Fatalf("reply = %q", reply) + } + raw, err := os.ReadFile(logPath) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(raw), `"--attachment", "`+attachmentPath+`"`) || !strings.Contains(string(raw), `"-p", "分析视频"`) { + t.Fatalf("qoder args missing attachment: %s", raw) + } +} + func TestQoderForwarderKeepsStreamJSONProcessAlive(t *testing.T) { t.Setenv("DWS_CONNECT_NO_INSTALL", "1") t.Setenv("DWS_AGENT_CMD", "") diff --git a/internal/helpers/connect_stream.go b/internal/helpers/connect_stream.go index ed25712b..0421c864 100644 --- a/internal/helpers/connect_stream.go +++ b/internal/helpers/connect_stream.go @@ -74,6 +74,44 @@ type streamingForwarder interface { forwardStream(ctx context.Context, convID, text string, onDelta func(string)) (string, error) } +// connectMediaAttachment is a DingTalk attachment that has already been +// authenticated and downloaded by the common Stream ingress. Keeping the +// attachment separate from the textual prompt lets multimodal backends pass +// the original bytes through their native protocol instead of asking the +// model to infer a local path from prose. +type connectMediaAttachment struct { + LocalPath string + FileName string + MediaType string +} + +// attachmentForwarder is implemented by backends with a native attachment +// transport (for example OpenCode file parts or Gemini inlineData). Backends +// without one still receive the absolute local path in the prompt and can use +// their read tool, preserving compatibility with custom agents. +type attachmentForwarder interface { + forwardWithAttachments(ctx context.Context, convID, text string, attachments []connectMediaAttachment) (string, error) +} + +type streamingAttachmentForwarder interface { + attachmentForwarder + canStream() bool + forwardStreamWithAttachments(ctx context.Context, convID, text string, attachments []connectMediaAttachment, onDelta func(string)) (string, error) +} + +func forwardConnectTurn(ctx context.Context, fwd forwarder, convID, prompt string, attachments []connectMediaAttachment, onDelta func(string)) (string, error) { + if af, ok := fwd.(streamingAttachmentForwarder); ok { + return af.forwardStreamWithAttachments(ctx, convID, prompt, attachments, onDelta) + } + if af, ok := fwd.(attachmentForwarder); ok { + return af.forwardWithAttachments(ctx, convID, prompt, attachments) + } + if sf, ok := fwd.(streamingForwarder); ok { + return sf.forwardStream(ctx, convID, prompt, onDelta) + } + return fwd.forward(ctx, convID, prompt) +} + // sessionResetter is an optional capability: a forwarder that can forget a // conversation's agent session, so a built-in /new or /clear command starts a // fresh context. Forwarders with per-conversation memory (Claude-family exec, @@ -339,6 +377,44 @@ func (f *execForwarder) forward(ctx context.Context, convID, text string) (strin return "", fmt.Errorf("本地 %s agent 调用失败：%s", f.name, truncateRunes(msg, 300)) } +// forwardWithAttachments grants the Claude-family CLIs read-only access to the +// exact directories that contain this turn's downloaded attachments. These +// agents otherwise run from an isolated scratch directory, so an absolute path +// in prose can still be rejected by their external-directory permission gate. +// The custom channel is intentionally left untouched because DWS cannot assume +// flags understood by an arbitrary user command. +func (f *execForwarder) forwardWithAttachments(ctx context.Context, convID, text string, attachments []connectMediaAttachment) (string, error) { + switch f.name { + case "claudecode", "codebuddy", "workbuddy": + default: + return f.forward(ctx, convID, text) + } + seen := make(map[string]struct{}) + var dirs []string + for _, attachment := range attachments { + path := strings.TrimSpace(attachment.LocalPath) + if path == "" { + continue + } + dir := filepath.Dir(path) + if _, exists := seen[dir]; exists { + continue + } + seen[dir] = struct{}{} + dirs = append(dirs, dir) + } + if len(dirs) == 0 { + return f.forward(ctx, convID, text) + } + clone := *f + clone.argv = append([]string{f.argv[0]}, "--allowedTools", "Read") + for _, dir := range dirs { + clone.argv = append(clone.argv, "--add-dir", dir) + } + clone.argv = append(clone.argv, f.argv[1:]...) + return clone.forward(ctx, convID, text) +} + // convSessions maps a DingTalk conversation to a stable agent session ID, so a // channel CLI with addressable sessions keeps multi-turn context per chat. // First message of a conversation mints a UUID and passes `--session-id ` @@ -699,10 +775,10 @@ var agentSpecs = map[string]agentSpec{ "workbuddy": {app: "WorkBuddy（自带 codebuddy）", bins: []string{"codebuddy"}, globs: []string{"/Applications/WorkBuddy.app/Contents/Resources/app.asar.unpacked/cli/bin/codebuddy"}, argvTail: []string{"--append-system-prompt", - "你叫「WorkBuddy 助手」，是钉钉群里的智能助手。无论被问到你是谁，都只能自称 WorkBuddy 助手，绝不能提到 CodeBuddy 这个名字。请用简洁自然的中文直接回答问题；不要使用任何工具，不要尝试读写文件或执行命令。", + "你叫「WorkBuddy 助手」，是钉钉群里的智能助手。无论被问到你是谁，都只能自称 WorkBuddy 助手，绝不能提到 CodeBuddy 这个名字。请用简洁自然的中文直接回答问题；不要主动使用工具、读写文件或执行命令。仅当用户消息明确附带了本地附件路径时，可以使用 Read 工具只读该附件，不得访问其它文件。", "-p"}, streamArgvTail: []string{"--append-system-prompt", - "你叫「WorkBuddy 助手」，是钉钉群里的智能助手。无论被问到你是谁，都只能自称 WorkBuddy 助手，绝不能提到 CodeBuddy 这个名字。请用简洁自然的中文直接回答问题；不要使用任何工具，不要尝试读写文件或执行命令。", + "你叫「WorkBuddy 助手」，是钉钉群里的智能助手。无论被问到你是谁，都只能自称 WorkBuddy 助手，绝不能提到 CodeBuddy 这个名字。请用简洁自然的中文直接回答问题；不要主动使用工具、读写文件或执行命令。仅当用户消息明确附带了本地附件路径时，可以使用 Read 工具只读该附件，不得访问其它文件。", "-p", "--output-format", "stream-json", "--include-partial-messages"}, streamParser: "cc", envFn: codebuddyEnv, hint: "https://www.codebuddy.cn/work/", modelFlag: "--model", ccSessions: true}, @@ -994,17 +1070,18 @@ type connectExtras struct { } type connectQueuedTurn struct { - convID string - text string - picCodes []string - fileInfo fileInboundInfo - webhook string - msgID string - msgType string - senderStaffID string - conversationID string - conversationType string - callbackData chatbot.BotCallbackDataModel + convID string + text string + picCodes []string + fileInfos []fileInboundInfo + chatRecordLookups []chatRecordLookup + webhook string + msgID string + msgType string + senderStaffID string + conversationID string + conversationType string + callbackData chatbot.BotCallbackDataModel } func mergeConnectQueuedTurns(turns []connectQueuedTurn) connectQueuedTurn { @@ -1027,16 +1104,12 @@ func mergeConnectQueuedTurns(turns []connectQueuedTurn) connectQueuedTurn { } merged.text = strings.Join(lines, "\n") merged.picCodes = nil + merged.fileInfos = nil + merged.chatRecordLookups = nil for i := range turns { merged.picCodes = append(merged.picCodes, turns[i].picCodes...) - } - if !merged.fileInfo.hasActionable() { - for i := len(turns) - 1; i >= 0; i-- { - if turns[i].fileInfo.hasActionable() { - merged.fileInfo = turns[i].fileInfo - break - } - } + merged.fileInfos = append(merged.fileInfos, turns[i].fileInfos...) + merged.chatRecordLookups = append(merged.chatRecordLookups, turns[i].chatRecordLookups...) } return merged } @@ -1061,11 +1134,13 @@ func connectTurnSummary(turn connectQueuedTurn) string { if len(turn.picCodes) > 0 { return "[图片]" } - if turn.fileInfo.hasActionable() { - if name := strings.TrimSpace(turn.fileInfo.FileName); name != "" { - return "[文件: " + name + "]" + if len(turn.fileInfos) > 0 { + if len(turn.fileInfos) == 1 { + if name := strings.TrimSpace(turn.fileInfos[0].FileName); name != "" { + return "[附件: " + name + "]" + } } - return "[文件]" + return fmt.Sprintf("[%d 个附件]", len(turn.fileInfos)) } return "[空消息]" } @@ -1118,41 +1193,40 @@ func runStreamConnector(ctx context.Context, channel, clientID, clientSecret str cli.RegisterChatBotCallbackRouter(func(_ context.Context, data *chatbot.BotCallbackDataModel) ([]byte, error) { text := strings.TrimSpace(data.Text.Content) msgtype := strings.TrimSpace(data.Msgtype) - // Picture messages carry no text — their payload is a downloadCode - // resolved to a local file in the forward goroutine below. - var picCodes []string - if strings.EqualFold(msgtype, "picture") { - if code := pictureDownloadCode(data.Content); code != "" { - picCodes = append(picCodes, code) + // Discover attachments by their locator fields, never by a msgtype + // allowlist. msgtype is only a classification hint for the agent prompt. + picCodes, fileInfos, unrecoverableCount := callbackInboundMedia(msgtype, data.Content) + var chatRecordLookups []chatRecordLookup + if unrecoverableCount > 0 { + indexes := chatRecordUnknownIndexes(data.Content) + if strings.TrimSpace(data.MsgId) != "" && len(indexes) > 0 { + chatRecordLookups = append(chatRecordLookups, chatRecordLookup{ + MsgID: strings.TrimSpace(data.MsgId), + UnknownIndexes: indexes, + }) + fmt.Fprintf(os.Stderr, "[connect][media] 转发记录中有 %d 条 unknownMsgType，将在 ACK 后补拉原始内容 (msgId=%s)\n", unrecoverableCount, data.MsgId) + } else { + fmt.Fprintf(os.Stderr, "[connect][media] 转发记录中有 %d 条 unknownMsgType，但缺少外层消息 ID，保留原始 JSON 降级处理\n", unrecoverableCount) } } - // A richText callback can mix text and inline picture nodes. Scan its - // content even when data.Text.Content is already populated; otherwise - // the text survives while every embedded picture silently disappears. - if strings.EqualFold(msgtype, "richText") { - picCodes = append(picCodes, richTextPictureDownloadCodes(data.Content)...) - } - // File callbacks come in two shapes: client-sent files carry a - // downloadCode; API-sent files (`dws chat message send --msg-type file - // --dentry-id --space-id`) carry dentryId + spaceId instead and have - // NO downloadCode. Both have to be recognisable or legit file messages - // get silently dropped below. - var fileInfo fileInboundInfo - if strings.EqualFold(msgtype, "file") { - fileInfo = parseFileInbound(data.Content) - } // Structured-text fallback: DingTalk leaves data.Text.Content blank on // markdown / richText callbacks (the body ships in data.Content). Without // this, `dws chat message send --group ... --text ...` — which defaults // to msgType=markdown — hits the drop branch below and the bot looks // dead to the sender. - if text == "" && len(picCodes) == 0 { + if text == "" { + // Forwarded records must keep their complete JSON, even if the outer + // msgtype is renamed or a title-like field could be extracted as text. + // Detect the record by payload shape rather than message type. + if hasChatRecordPayload(data.Content) { + text = rawCallbackPrompt(msgtype, data.Content) + } // interactiveCard (a bot @-mentioning this bot) nests the body in // content.cardContent and carries the mention as its own leading // leaf; the leaf-aware extractor drops it so the agent gets the - // clean instruction. Other structured-text shapes use the generic - // extractor. - if strings.EqualFold(msgtype, "interactiveCard") { + // clean instruction. Detection is based on the payload shape so a + // renamed/new type is handled identically. + if text == "" { text = extractInteractiveCardText(data.Content) } if text == "" { @@ -1160,13 +1234,15 @@ func runStreamConnector(ctx context.Context, channel, clientID, clientSecret str text = fallback } } + if text == "" { + text = rawCallbackPrompt(msgtype, data.Content) + } } - if (text == "" && len(picCodes) == 0 && !fileInfo.hasActionable()) || data.SessionWebhook == "" { - // Observability: silent drops are the #1 reason a working connector - // looks dead. Log msgtype + a payload summary so an unhandled shape - // (e.g. new-style file callback without downloadCode) shows up in - // stderr instead of being invisible. - fmt.Fprintf(os.Stderr, "[connect] 丢弃消息 msgtype=%q staffId=%s convId=%s msgId=%s content=%s (无正文/图片/可下载文件或 sessionWebhook 为空)\n", + if data.SessionWebhook == "" { + // A session webhook is required for the fallback reply path. Message + // payload shape is deliberately not filtered here: unknown and complex + // types are forwarded as raw JSON for the backend model to interpret. + fmt.Fprintf(os.Stderr, "[connect] 丢弃消息 msgtype=%q staffId=%s convId=%s msgId=%s content=%s (sessionWebhook 为空，无法回复)\n", msgtype, data.SenderStaffId, data.ConversationId, data.MsgId, summarizeContent(data.Content)) return []byte(""), nil } @@ -1199,8 +1275,8 @@ func runStreamConnector(ctx context.Context, channel, clientID, clientSecret str shown := text if shown == "" && len(picCodes) > 0 { shown = "[图片]" - } else if shown == "" && fileInfo.hasActionable() { - shown = "[文件: " + fileInfo.FileName + "]" + } else if shown == "" && len(fileInfos) > 0 { + shown = connectTurnSummary(connectQueuedTurn{fileInfos: fileInfos}) } fmt.Fprintf(os.Stderr, "[connect] 收到 @%s: %s (convType=%s convId=%s staffId=%s msgId=%s)\n", sender, truncateRunes(shown, 80), data.ConversationType, data.ConversationId, data.SenderStaffId, data.MsgId) @@ -1217,17 +1293,18 @@ func runStreamConnector(ctx context.Context, channel, clientID, clientSecret str } msgID := strings.TrimSpace(data.MsgId) turn := connectQueuedTurn{ - convID: convID, - text: text, - picCodes: picCodes, - fileInfo: fileInfo, - webhook: webhook, - msgID: msgID, - msgType: msgtype, - senderStaffID: strings.TrimSpace(data.SenderStaffId), - conversationID: strings.TrimSpace(data.ConversationId), - conversationType: strings.TrimSpace(data.ConversationType), - callbackData: *data, + convID: convID, + text: text, + picCodes: picCodes, + fileInfos: fileInfos, + chatRecordLookups: chatRecordLookups, + webhook: webhook, + msgID: msgID, + msgType: msgtype, + senderStaffID: strings.TrimSpace(data.SenderStaffId), + conversationID: strings.TrimSpace(data.ConversationId), + conversationType: strings.TrimSpace(data.ConversationType), + callbackData: *data, } // Same-conversation agent calls never run in parallel; messages received // while a turn is running are merged into one pending follow-up instead @@ -1239,7 +1316,8 @@ func runStreamConnector(ctx context.Context, channel, clientID, clientSecret str } text := turn.text picCodes := turn.picCodes - fileInfo := turn.fileInfo + fileInfos := turn.fileInfos + chatRecordLookups := turn.chatRecordLookups webhook := turn.webhook convID := turn.convID msgID := turn.msgID @@ -1292,6 +1370,28 @@ func runStreamConnector(ctx context.Context, channel, clientID, clientSecret str // Assemble the forwarded prompt: resolve an attached picture (the // top Q&A inbound is an error screenshot), then knowledge-augment. prompt := text + for _, lookup := range chatRecordLookups { + lookupCtx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + enrichment, lookupErr := recoverChatRecordUnknowns(lookupCtx, lookup, callMCPToolReturnTextOnServer) + cancel() + if lookupErr != nil { + fmt.Fprintf(os.Stderr, "[connect][media] unknownMsgType 补拉失败，保留原始 JSON (msgId=%s): %v\n", lookup.MsgID, lookupErr) + continue + } + if strings.TrimSpace(enrichment.Prompt) != "" { + if strings.TrimSpace(prompt) == "" { + prompt = enrichment.Prompt + } else { + prompt += "\n\n" + enrichment.Prompt + } + } + fileInfos = append(fileInfos, enrichment.Files...) + fmt.Fprintf(os.Stderr, "[connect][media] unknownMsgType 补拉完成: 原始附件=%d 未定位=%d (msgId=%s)\n", len(enrichment.Files), enrichment.MissingCount, lookup.MsgID) + if enrichment.MissingCount > 0 { + prompt += fmt.Sprintf("\n（其中 %d 个转发附件仍未能定位原始文件，请明确告知用户未读取到这些附件。）", enrichment.MissingCount) + } + } + var attachments []connectMediaAttachment for i, picCode := range picCodes { localPath, derr := mediaCli.downloadMessageFile(context.Background(), clientID, picCode) if derr != nil { @@ -1308,23 +1408,58 @@ func runStreamConnector(ctx context.Context, channel, clientID, clientSecret str } else { prompt += "\n（用户同时附了一张图片，本地路径 " + localPath + "，请结合图片内容回答。）" } + attachments = append(attachments, connectMediaAttachment{ + LocalPath: localPath, + FileName: filepath.Base(localPath), + MediaType: "image", + }) } - if fileInfo.hasActionable() { + for i, fileInfo := range fileInfos { + if !fileInfo.hasActionable() { + continue + } fileName := fileInfo.FileName + mediaType := inboundMediaType(fileInfo.MediaType) + mediaLabel := "文件" + successPrompt := "请读取文件内容并回答" + switch mediaType { + case "image": + mediaLabel = "图片" + successPrompt = "请查看图片内容并回答" + case "audio": + mediaLabel = "语音" + successPrompt = "请听取或转写语音内容并回答" + case "video": + mediaLabel = "视频" + successPrompt = "请查看并分析视频内容后回答" + } var localPath string var derr error if fileInfo.DownloadCode != "" { - localPath, derr = mediaCli.downloadMessageFile(context.Background(), clientID, fileInfo.DownloadCode) + localPath, derr = mediaCli.downloadMessageFileNamed(context.Background(), clientID, fileInfo.DownloadCode, fileName) if derr != nil { - fmt.Fprintf(os.Stderr, "[connect][media] 文件下载失败: %v\n", derr) + fmt.Fprintf(os.Stderr, "[connect][media] 第 %d 个%s下载失败: %v\n", i+1, mediaLabel, derr) + } + } else if fileInfo.MediaID != "" || fileInfo.FileID != "" { + downloadCtx, cancel := context.WithTimeout(context.Background(), mediaDownloadTimeout) + localPath, derr = mediaCli.downloadRecoveredChatRecordFile(downloadCtx, fileInfo) + cancel() + if derr != nil { + fmt.Fprintf(os.Stderr, "[connect][media] 第 %d 个转发%s原始内容下载失败: %v\n", i+1, mediaLabel, derr) } } switch { case localPath != "": + fmt.Fprintf(os.Stderr, "[connect][media] 第 %d 个%s已完整下载: %s\n", i+1, mediaLabel, localPath) + attachments = append(attachments, connectMediaAttachment{ + LocalPath: localPath, + FileName: fileName, + MediaType: mediaType, + }) if prompt == "" { - prompt = "用户发来一个文件「" + fileName + "」（本地路径 " + localPath + "），请读取文件内容并回答。" + prompt = "用户发来一个" + mediaLabel + "「" + fileName + "」（本地路径 " + localPath + "），" + successPrompt + "。" } else { - prompt = prompt + "\n（用户同时附了一个文件「" + fileName + "」，本地路径 " + localPath + "，请结合文件内容回答。）" + prompt += "\n（用户同时附了一个" + mediaLabel + "「" + fileName + "」，本地路径 " + localPath + "，" + successPrompt + "。）" } case fileInfo.DentryID != 0 && fileInfo.SpaceID != 0: // API-sent file: resolve via storage API (userId→unionId, @@ -1339,10 +1474,16 @@ func runStreamConnector(ctx context.Context, channel, clientID, clientSecret str localPath = dp } if localPath != "" { + fmt.Fprintf(os.Stderr, "[connect][media] 第 %d 个%s已完整下载: %s\n", i+1, mediaLabel, localPath) + attachments = append(attachments, connectMediaAttachment{ + LocalPath: localPath, + FileName: fileName, + MediaType: mediaType, + }) if prompt == "" { - prompt = "用户发来一个文件「" + fileName + "」（本地路径 " + localPath + "），请读取文件内容并回答。" + prompt = "用户发来一个" + mediaLabel + "「" + fileName + "」（本地路径 " + localPath + "），" + successPrompt + "。" } else { - prompt = prompt + "\n（用户同时附了一个文件「" + fileName + "」，本地路径 " + localPath + "，请结合文件内容回答。）" + prompt += "\n（用户同时附了一个" + mediaLabel + "「" + fileName + "」，本地路径 " + localPath + "，" + successPrompt + "。）" } } else { meta := fmt.Sprintf("文件名「%s」，dentryId=%d，spaceId=%d", fileName, fileInfo.DentryID, fileInfo.SpaceID) @@ -1359,11 +1500,22 @@ func runStreamConnector(ctx context.Context, channel, clientID, clientSecret str } } default: + failure := "用户发来一个" + mediaLabel + "「" + fileName + "」，但原始内容下载失败。请明确告知用户该附件未能读取，建议重新发送或补充文字描述。" if prompt == "" { - prompt = "（用户发来一个文件「" + fileName + "」，但文件下载失败了。请告知用户文件没收到，建议重新发送。）" + prompt = "（" + failure + "）" + } else { + prompt += "\n（" + failure + "）" } } } + originalAttachments := append([]connectMediaAttachment(nil), attachments...) + defer cleanupConnectMediaAttachments(originalAttachments) + if _, isOpenCode := fwd.(*opencodeForwarder); isOpenCode { + prepareCtx, cancel := context.WithTimeout(context.Background(), mediaDownloadTimeout) + prompt, attachments = prepareOpenCodeAttachments(prepareCtx, prompt, attachments) + cancel() + } + defer cleanupConnectMediaAttachments(attachments) if extras.kb != nil { prompt = extras.kb.augment(prompt) } @@ -1421,13 +1573,7 @@ func runStreamConnector(ctx context.Context, channel, clientID, clientSecret str } } - var reply string - var err error - if streamable { - reply, err = sf.forwardStream(context.Background(), convID, prompt, onDelta) - } else { - reply, err = fwd.forward(context.Background(), convID, prompt) - } + reply, err := forwardConnectTurn(context.Background(), fwd, convID, prompt, attachments, onDelta) if err != nil { fmt.Fprintf(os.Stderr, "[connect] 转发失败 (%s, 耗时 %s): %v\n", channel, time.Since(started).Round(time.Millisecond), err) if errors.Is(err, context.DeadlineExceeded) { diff --git a/internal/helpers/connect_stream_test.go b/internal/helpers/connect_stream_test.go index 02bd73ca..c06527a8 100644 --- a/internal/helpers/connect_stream_test.go +++ b/internal/helpers/connect_stream_test.go @@ -5,13 +5,97 @@ package helpers import ( + "context" "os" "path/filepath" + "reflect" "strings" "testing" "time" ) +type attachmentRecordingForwarder struct { + attachments []connectMediaAttachment +} + +func (f *attachmentRecordingForwarder) label() string { return "recording" } +func (f *attachmentRecordingForwarder) forward(context.Context, string, string) (string, error) { + return "text-only", nil +} +func (f *attachmentRecordingForwarder) forwardWithAttachments(_ context.Context, _ string, _ string, attachments []connectMediaAttachment) (string, error) { + f.attachments = append([]connectMediaAttachment(nil), attachments...) + return "with-attachments", nil +} + +func TestForwardConnectTurnPreservesAttachmentsForCapableAgent(t *testing.T) { + fwd := &attachmentRecordingForwarder{} + want := []connectMediaAttachment{{LocalPath: "/tmp/original.mov", FileName: "original.mov", MediaType: "video"}} + reply, err := forwardConnectTurn(context.Background(), fwd, "conv", "prompt", want, nil) + if err != nil { + t.Fatal(err) + } + if reply != "with-attachments" || len(fwd.attachments) != 1 || fwd.attachments[0] != want[0] { + t.Fatalf("reply=%q attachments=%#v", reply, fwd.attachments) + } +} + +func TestExecForwarderAllowsOnlyAttachmentDirectory(t *testing.T) { + dir := t.TempDir() + logPath := filepath.Join(dir, "args.log") + stub := filepath.Join(dir, "agent") + script := "#!/bin/sh\nprintf '%s\\n' \"$@\" > \"$DWS_ARGS_LOG\"\nprintf 'ok\\n'\n" + if err := os.WriteFile(stub, []byte(script), 0o755); err != nil { + t.Fatal(err) + } + attachmentDir := t.TempDir() + attachmentPath := filepath.Join(attachmentDir, "report.md") + if err := os.WriteFile(attachmentPath, []byte("body"), 0o600); err != nil { + t.Fatal(err) + } + f := &execForwarder{name: "workbuddy", argv: []string{stub, "-p"}, env: []string{"DWS_ARGS_LOG=" + logPath}, timeout: time.Second} + reply, err := f.forwardWithAttachments(context.Background(), "conv", "read it", []connectMediaAttachment{{LocalPath: attachmentPath, FileName: "report.md", MediaType: "file"}}) + if err != nil { + t.Fatal(err) + } + if reply != "ok" { + t.Fatalf("reply = %q", reply) + } + raw, err := os.ReadFile(logPath) + if err != nil { + t.Fatal(err) + } + args := string(raw) + for _, want := range []string{"--allowedTools", "Read", "--add-dir", attachmentDir, "-p", "read it"} { + if !strings.Contains(args, want+"\n") { + t.Fatalf("args missing %q:\n%s", want, args) + } + } +} + +func TestEveryStreamBridgeAgentHasAttachmentDeliveryPath(t *testing.T) { + implementations := map[string]any{ + "claudecode": (*execForwarder)(nil), + "codebuddy": (*execForwarder)(nil), + "workbuddy": (*execForwarder)(nil), + "custom": (*execForwarder)(nil), + "qoder": (*qoderStreamForwarder)(nil), + "qoderwork": (*qoderStreamForwarder)(nil), + "codex": (*codexAppServerForwarder)(nil), + "opencode": (*opencodeForwarder)(nil), + "gemini": (*geminiAPIForwarder)(nil), + } + for channel := range agentSpecs { + impl, ok := implementations[channel] + if !ok { + t.Errorf("agent channel %q has no declared attachment delivery path", channel) + continue + } + if _, ok := impl.(attachmentForwarder); !ok { + t.Errorf("agent channel %q implementation %T is not attachment-aware", channel, impl) + } + } +} + // TestBrandReply covers the qoderwork identity rewrite using the exact replies // captured from a real qodercli (QoderWork.app) headless run. func TestBrandReply(t *testing.T) { @@ -200,6 +284,48 @@ func TestMergeConnectQueuedTurnsPreservesAllPictures(t *testing.T) { } } +func TestMergeConnectQueuedTurnsPreservesAllAttachments(t *testing.T) { + merged := mergeConnectQueuedTurns([]connectQueuedTurn{ + { + convID: "conv-1", + text: "第一批", + fileInfos: []fileInboundInfo{{DownloadCode: "audio-1", FileName: "语音消息", MediaType: "audio"}}, + msgID: "m1", + }, + { + convID: "conv-1", + text: "第二批", + fileInfos: []fileInboundInfo{ + {DownloadCode: "video-1", FileName: "demo.mov", MediaType: "video"}, + {DownloadCode: "file-1", FileName: "report.md", MediaType: "file"}, + }, + msgID: "m2", + }, + }) + if len(merged.fileInfos) != 3 { + t.Fatalf("merged fileInfos = %#v, want all three attachments", merged.fileInfos) + } + for i, want := range []string{"audio-1", "video-1", "file-1"} { + if merged.fileInfos[i].DownloadCode != want { + t.Fatalf("merged fileInfos[%d] = %#v, want code %q", i, merged.fileInfos[i], want) + } + } +} + +func TestMergeConnectQueuedTurnsKeepsEveryChatRecordLookup(t *testing.T) { + merged := mergeConnectQueuedTurns([]connectQueuedTurn{ + {convID: "conv-1", text: "first", chatRecordLookups: []chatRecordLookup{{MsgID: "outer-1", UnknownIndexes: []int{1}}}}, + {convID: "conv-1", text: "second", chatRecordLookups: []chatRecordLookup{{MsgID: "outer-2", UnknownIndexes: []int{0, 2}}}}, + }) + want := []chatRecordLookup{ + {MsgID: "outer-1", UnknownIndexes: []int{1}}, + {MsgID: "outer-2", UnknownIndexes: []int{0, 2}}, + } + if !reflect.DeepEqual(merged.chatRecordLookups, want) { + t.Fatalf("chatRecordLookups = %#v, want %#v", merged.chatRecordLookups, want) + } +} + func TestMergeConnectQueuedTurnsKeepsControlMessagesStandalone(t *testing.T) { for _, text := range []string{"/clear", "同意", "拒绝", "重试"} { merged := mergeConnectQueuedTurns([]connectQueuedTurn{