From 8f8d28880d41e2bdd8d1e2398423f62c170da09d Mon Sep 17 00:00:00 2001 From: leookun Date: Wed, 12 Aug 2026 01:10:14 +0800 Subject: [PATCH] feat(forwarder): persist read images by content hash --- internal/backend/agent/bridge/exec/bridge.go | 81 +++++- .../bridge/exec/bridge_read_image_test.go | 113 ++++++-- internal/backend/forwarder/compiler.go | 13 +- .../backend/forwarder/content_blob_store.go | 111 ++++++++ .../forwarder/content_blob_store_test.go | 41 +++ internal/backend/forwarder/projector.go | 55 ---- .../forwarder/projector_read_image_test.go | 241 ++++++++++++------ .../backend/forwarder/provider_read_images.go | 174 +++++++++++++ internal/backend/forwarder/service.go | 24 +- 9 files changed, 685 insertions(+), 168 deletions(-) create mode 100644 internal/backend/forwarder/content_blob_store.go create mode 100644 internal/backend/forwarder/content_blob_store_test.go create mode 100644 internal/backend/forwarder/provider_read_images.go diff --git a/internal/backend/agent/bridge/exec/bridge.go b/internal/backend/agent/bridge/exec/bridge.go index 32a8d47..b4701f9 100644 --- a/internal/backend/agent/bridge/exec/bridge.go +++ b/internal/backend/agent/bridge/exec/bridge.go @@ -2,8 +2,14 @@ package execbridge import ( + "bytes" + "crypto/sha256" "encoding/json" "fmt" + "image" + _ "image/gif" + _ "image/jpeg" + _ "image/png" "net/http" "strings" "sync/atomic" @@ -32,10 +38,18 @@ type ExecApplyResult struct { ToolResultPayload string // ToolCall 保存可用于发 ToolCallCompletedUpdate 的工具调用对象；当前仅对支持 ToolCall 的执行型工具可用。 ToolCall *agentv1.ToolCall + // ContentBlobs 保存需要在提交 history 前写入内容寻址存储的二进制内容。 + ContentBlobs []ContentBlob // ExecuteHookResponse 保存 execute hook 的结构化响应。 ExecuteHookResponse *agentv1.ExecuteHookResponse } +// ContentBlob 表示由内容哈希稳定寻址的执行结果二进制数据。 +type ContentBlob struct { + ID []byte + Data []byte +} + // OpenExecContext 表示执行桥打开请求时需要的最小上下文。 type OpenExecContext struct { ConversationID string @@ -146,6 +160,9 @@ func (bridge *Bridge) ApplyExecClientMessage(msg *agentv1.ExecClientMessage, pen readResult := normalizeReadResultForModel(msg.GetReadResult()) result.ToolResultPayload = summarizeReadResult(readResult) result.ToolCall = buildReadCompletedToolCall(pending.ToolCallID, pending.ArgsJSON, readResult) + if contentBlob, ok := readImageContentBlob(readResult); ok { + result.ContentBlobs = []ContentBlob{contentBlob} + } result.IsTerminal = true return result, nil case "write": @@ -2286,16 +2303,62 @@ func buildReadMcpResourceCompletedToolCall(argsJSON []byte, result *agentv1.Read } } -func isSupportedReadImage(data []byte) bool { +func supportedReadImageMIMEType(data []byte) string { if len(data) == 0 { - return false + return "" } - switch strings.ToLower(strings.TrimSpace(http.DetectContentType(data))) { - case "image/png", "image/jpeg", "image/gif", "image/webp": - return true - default: - return false + detected := strings.ToLower(strings.TrimSpace(http.DetectContentType(data))) + configuration, format, err := image.DecodeConfig(bytes.NewReader(data)) + if err != nil || configuration.Width <= 0 || configuration.Height <= 0 { + return "" } + switch strings.ToLower(strings.TrimSpace(format)) { + case "png": + if detected == "image/png" { + return detected + } + case "jpeg": + if detected == "image/jpeg" { + return detected + } + case "gif": + if detected == "image/gif" { + return detected + } + } + return "" +} + +func readImageContentBlob(result *agentv1.ReadResult) (ContentBlob, bool) { + success := result.GetSuccess() + if success == nil { + return ContentBlob{}, false + } + data := success.GetData() + if supportedReadImageMIMEType(data) == "" { + return ContentBlob{}, false + } + digest := sha256.Sum256(data) + return ContentBlob{ + ID: append([]byte(nil), digest[:]...), + Data: append([]byte(nil), data...), + }, true +} + +func readImageBlobID(data []byte) ([]byte, bool) { + if supportedReadImageMIMEType(data) == "" { + return nil, false + } + digest := sha256.Sum256(data) + return append([]byte(nil), digest[:]...), true +} + +func readImageDataBlobOutput(data []byte) *agentv1.ReadToolSuccess_DataBlobId { + blobID, ok := readImageBlobID(data) + if !ok { + return nil + } + return &agentv1.ReadToolSuccess_DataBlobId{DataBlobId: blobID} } // convertReadResultToReadToolResult 把 `ReadResult` 映射为 `ReadToolResult`。 @@ -2331,8 +2394,8 @@ func convertReadResultToReadToolResult(result *agentv1.ReadResult) *agentv1.Read if content != "" { toolSuccess.Output = &agentv1.ReadToolSuccess_Content{Content: content} } else if len(data) > 0 { - if isSupportedReadImage(data) { - toolSuccess.Output = &agentv1.ReadToolSuccess_Data{Data: append([]byte(nil), data...)} + if imageOutput := readImageDataBlobOutput(data); imageOutput != nil { + toolSuccess.Output = imageOutput } else if len(data) > readReplayBinaryLimit { toolSuccess.ExceededLimit = true toolSuccess.Output = &agentv1.ReadToolSuccess_Content{ diff --git a/internal/backend/agent/bridge/exec/bridge_read_image_test.go b/internal/backend/agent/bridge/exec/bridge_read_image_test.go index 966eca1..6c8727e 100644 --- a/internal/backend/agent/bridge/exec/bridge_read_image_test.go +++ b/internal/backend/agent/bridge/exec/bridge_read_image_test.go @@ -2,31 +2,94 @@ package execbridge import ( "bytes" + "crypto/sha256" + "image" + "image/color" + "image/png" "strings" "testing" "cursor/gen/agentv1" + runtimecore "cursor/internal/backend/agent/core" ) -func TestConvertReadResultPreservesLargeImagesOnly(t *testing.T) { - largePNG := append([]byte("\x89PNG\r\n\x1a\n"), bytes.Repeat([]byte{0}, readReplayBinaryLimit)...) - imageResult := convertReadResultToReadToolResult(&agentv1.ReadResult{ +func TestApplyExecClientMessageReturnsContentAddressedReadImage(t *testing.T) { + imageData := validReadTestPNG(t) + wantBlobID := sha256.Sum256(imageData) + result, err := NewBridge().ApplyExecClientMessage(&agentv1.ExecClientMessage{ + Message: &agentv1.ExecClientMessage_ReadResult{ + ReadResult: &agentv1.ReadResult{ + Result: &agentv1.ReadResult_Success{ + Success: &agentv1.ReadSuccess{ + Path: "diagram.png", + FileSize: int64(len(imageData)), + OutputBlobId: append([]byte(nil), wantBlobID[:]...), + Output: &agentv1.ReadSuccess_Data{Data: imageData}, + }, + }, + }, + }, + }, runtimecore.PendingExec{ + ExecKind: "read", + ToolCallID: "call-1", + ArgsJSON: []byte(`{"path":"diagram.png"}`), + }) + if err != nil { + t.Fatalf("ApplyExecClientMessage() error = %v", err) + } + if len(result.ContentBlobs) != 1 { + t.Fatalf("content blob count = %d, want 1", len(result.ContentBlobs)) + } + if !bytes.Equal(result.ContentBlobs[0].ID, wantBlobID[:]) || !bytes.Equal(result.ContentBlobs[0].Data, imageData) { + t.Fatalf("content blob = %#v", result.ContentBlobs[0]) + } + readSuccess := result.ToolCall.GetReadToolCall().GetResult().GetSuccess() + if readSuccess == nil { + t.Fatal("read tool result is not successful") + } + if !bytes.Equal(readSuccess.GetDataBlobId(), wantBlobID[:]) { + t.Fatalf("data_blob_id = %x, want %x", readSuccess.GetDataBlobId(), wantBlobID) + } + if len(readSuccess.GetData()) != 0 { + t.Fatalf("read tool result retained %d image bytes", len(readSuccess.GetData())) + } +} + +func TestApplyExecClientMessageUsesComputedImageBlobID(t *testing.T) { + imageData := validReadTestPNG(t) + wantBlobID := sha256.Sum256(imageData) + result, err := NewBridge().ApplyExecClientMessage(&agentv1.ExecClientMessage{ + Message: &agentv1.ExecClientMessage_ReadResult{ + ReadResult: &agentv1.ReadResult{ + Result: &agentv1.ReadResult_Success{ + Success: &agentv1.ReadSuccess{ + Path: "diagram.png", + OutputBlobId: bytes.Repeat([]byte{0xff}, sha256.Size), + Output: &agentv1.ReadSuccess_Data{Data: imageData}, + }, + }, + }, + }, + }, runtimecore.PendingExec{ExecKind: "read", ToolCallID: "call-1"}) + if err != nil { + t.Fatalf("ApplyExecClientMessage() error = %v", err) + } + if !bytes.Equal(result.ContentBlobs[0].ID, wantBlobID[:]) { + t.Fatalf("content blob id = %x, want computed %x", result.ContentBlobs[0].ID, wantBlobID) + } +} + +func TestConvertReadResultKeepsTextAndLimitsUnsupportedBinary(t *testing.T) { + textResult := convertReadResultToReadToolResult(&agentv1.ReadResult{ Result: &agentv1.ReadResult_Success{ Success: &agentv1.ReadSuccess{ - Path: "image.png", - Output: &agentv1.ReadSuccess_Data{Data: largePNG}, + Path: "notes.txt", + Output: &agentv1.ReadSuccess_Content{Content: "hello"}, }, }, }) - imageSuccess := imageResult.GetSuccess() - if imageSuccess == nil { - t.Fatal("large image result is not successful") - } - if !bytes.Equal(imageSuccess.GetData(), largePNG) { - t.Fatalf("large image data bytes = %d, want %d", len(imageSuccess.GetData()), len(largePNG)) - } - if imageSuccess.GetExceededLimit() { - t.Fatal("large image unexpectedly marked as exceeded limit") + if got := textResult.GetSuccess().GetContent(); got != "hello" { + t.Fatalf("text read content = %q, want hello", got) } largeBinary := bytes.Repeat([]byte{0xff}, readReplayBinaryLimit+1) @@ -39,16 +102,24 @@ func TestConvertReadResultPreservesLargeImagesOnly(t *testing.T) { }, }) binarySuccess := binaryResult.GetSuccess() - if binarySuccess == nil { - t.Fatal("large binary result is not successful") + if binarySuccess == nil || !binarySuccess.GetExceededLimit() { + t.Fatal("large non-image binary was not limited") } - if !binarySuccess.GetExceededLimit() { - t.Fatal("large non-image binary was not marked as exceeded limit") - } - if binarySuccess.GetData() != nil { - t.Fatal("large non-image binary data was retained") + if binarySuccess.GetData() != nil || binarySuccess.GetDataBlobId() != nil { + t.Fatal("large non-image binary was retained") } if !strings.Contains(binarySuccess.GetContent(), "Read binary data") { t.Fatalf("large binary fallback = %q", binarySuccess.GetContent()) } } + +func validReadTestPNG(t *testing.T) []byte { + t.Helper() + value := image.NewRGBA(image.Rect(0, 0, 2, 2)) + value.Set(0, 0, color.RGBA{R: 0x44, G: 0x88, B: 0xcc, A: 0xff}) + var encoded bytes.Buffer + if err := png.Encode(&encoded, value); err != nil { + t.Fatalf("encode test png: %v", err) + } + return encoded.Bytes() +} diff --git a/internal/backend/forwarder/compiler.go b/internal/backend/forwarder/compiler.go index 93a5608..b6f5a02 100644 --- a/internal/backend/forwarder/compiler.go +++ b/internal/backend/forwarder/compiler.go @@ -20,16 +20,21 @@ type DefaultPromptCompiler struct { catalog ToolCatalog reminders ReminderInjector rules *UserRuleStore + blobs contentBlobReader } // NewPromptCompiler 创建默认 prompt 编译器。 -func NewPromptCompiler(projector *HistoryProjector, catalog ToolCatalog, reminders ReminderInjector, rules *UserRuleStore) *DefaultPromptCompiler { - return &DefaultPromptCompiler{ +func NewPromptCompiler(projector *HistoryProjector, catalog ToolCatalog, reminders ReminderInjector, rules *UserRuleStore, blobReaders ...contentBlobReader) *DefaultPromptCompiler { + compiler := &DefaultPromptCompiler{ projector: projector, catalog: catalog, reminders: reminders, rules: rules, } + if len(blobReaders) > 0 { + compiler.blobs = blobReaders[0] + } + return compiler } // Compile 生成当前 turn 应发送给 provider 的消息和工具集合。 @@ -86,6 +91,10 @@ func (compiler *DefaultPromptCompiler) Compile(conversation *ConversationFile, m if err != nil { return CompiledConversation{}, err } + replayMessages, err = enrichProviderReadImages(replayMessages, conversation, compiler.blobs) + if err != nil { + return CompiledConversation{}, err + } messages = append(messages, replayMessages...) return CompiledConversation{ Mode: normalizedMode, diff --git a/internal/backend/forwarder/content_blob_store.go b/internal/backend/forwarder/content_blob_store.go new file mode 100644 index 0000000..0e5351e --- /dev/null +++ b/internal/backend/forwarder/content_blob_store.go @@ -0,0 +1,111 @@ +// content_blob_store.go 负责持久化 history 引用的内容寻址二进制数据。 +package forwarder + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "fmt" + "os" + "path/filepath" + "strings" +) + +const contentBlobDirectoryName = ".blobs" + +// ContentBlobStore 使用 SHA-256 内容哈希保存不可变二进制数据。 +type ContentBlobStore struct { + root string +} + +// NewContentBlobStore 创建独立于 context.json 和 checkpoint 的内容寻址存储。 +func NewContentBlobStore(historyRoot string) *ContentBlobStore { + historyRoot = strings.TrimSpace(historyRoot) + if historyRoot == "" { + return &ContentBlobStore{} + } + return &ContentBlobStore{root: filepath.Join(historyRoot, contentBlobDirectoryName, "sha256")} +} + +// Put 校验内容哈希并幂等保存数据。 +func (store *ContentBlobStore) Put(id []byte, data []byte) error { + if store == nil || strings.TrimSpace(store.root) == "" { + return fmt.Errorf("content blob store is not initialized") + } + normalizedID, err := normalizeContentBlobID(id) + if err != nil { + return err + } + digest := sha256.Sum256(data) + if !bytes.Equal(normalizedID, digest[:]) { + return fmt.Errorf("content blob id does not match payload sha256") + } + path := store.blobPath(normalizedID) + if existing, err := store.Get(normalizedID); err == nil { + if bytes.Equal(existing, data) { + return nil + } + return fmt.Errorf("content blob payload conflicts with existing id") + } else if !os.IsNotExist(err) { + return err + } + if err := os.MkdirAll(store.root, 0o700); err != nil { + return fmt.Errorf("create content blob directory: %w", err) + } + temporary, err := os.CreateTemp(store.root, ".blob-*") + if err != nil { + return fmt.Errorf("create content blob temporary file: %w", err) + } + temporaryPath := temporary.Name() + defer os.Remove(temporaryPath) + if err := temporary.Chmod(0o600); err != nil { + _ = temporary.Close() + return fmt.Errorf("set content blob permissions: %w", err) + } + if _, err := temporary.Write(data); err != nil { + _ = temporary.Close() + return fmt.Errorf("write content blob: %w", err) + } + if err := temporary.Sync(); err != nil { + _ = temporary.Close() + return fmt.Errorf("sync content blob: %w", err) + } + if err := temporary.Close(); err != nil { + return fmt.Errorf("close content blob: %w", err) + } + if err := os.Rename(temporaryPath, path); err != nil { + return fmt.Errorf("commit content blob: %w", err) + } + return nil +} + +// Get 读取内容并再次校验哈希，避免损坏数据进入模型请求。 +func (store *ContentBlobStore) Get(id []byte) ([]byte, error) { + if store == nil || strings.TrimSpace(store.root) == "" { + return nil, fmt.Errorf("content blob store is not initialized") + } + normalizedID, err := normalizeContentBlobID(id) + if err != nil { + return nil, err + } + data, err := os.ReadFile(store.blobPath(normalizedID)) + if err != nil { + return nil, err + } + digest := sha256.Sum256(data) + if !bytes.Equal(normalizedID, digest[:]) { + return nil, fmt.Errorf("content blob sha256 verification failed") + } + return append([]byte(nil), data...), nil +} + +func (store *ContentBlobStore) blobPath(id []byte) string { + return filepath.Join(store.root, hex.EncodeToString(id)) +} + +func normalizeContentBlobID(id []byte) ([]byte, error) { + if len(id) != sha256.Size { + return nil, fmt.Errorf("content blob id must be %d bytes", sha256.Size) + } + return append([]byte(nil), id...), nil +} diff --git a/internal/backend/forwarder/content_blob_store_test.go b/internal/backend/forwarder/content_blob_store_test.go new file mode 100644 index 0000000..5e4cb7a --- /dev/null +++ b/internal/backend/forwarder/content_blob_store_test.go @@ -0,0 +1,41 @@ +package forwarder + +import ( + "bytes" + "crypto/sha256" + "testing" +) + +func TestContentBlobStorePutGetIsIdempotent(t *testing.T) { + store := NewContentBlobStore(t.TempDir()) + data := []byte("stable blob bytes") + id := sha256.Sum256(data) + if err := store.Put(id[:], data); err != nil { + t.Fatalf("first Put() error = %v", err) + } + if err := store.Put(id[:], append([]byte(nil), data...)); err != nil { + t.Fatalf("second Put() error = %v", err) + } + got, err := store.Get(id[:]) + if err != nil { + t.Fatalf("Get() error = %v", err) + } + if !bytes.Equal(got, data) { + t.Fatalf("Get() = %q, want %q", got, data) + } + got[0] ^= 0xff + again, err := store.Get(id[:]) + if err != nil { + t.Fatalf("second Get() error = %v", err) + } + if !bytes.Equal(again, data) { + t.Fatalf("stored data was mutated: %q", again) + } +} + +func TestContentBlobStoreRejectsMismatchedID(t *testing.T) { + store := NewContentBlobStore(t.TempDir()) + if err := store.Put(bytes.Repeat([]byte{0xff}, sha256.Size), []byte("payload")); err == nil { + t.Fatal("Put() accepted mismatched content id") + } +} diff --git a/internal/backend/forwarder/projector.go b/internal/backend/forwarder/projector.go index 7415afe..e310780 100644 --- a/internal/backend/forwarder/projector.go +++ b/internal/backend/forwarder/projector.go @@ -5,7 +5,6 @@ import ( "crypto/sha256" "encoding/json" "fmt" - "net/http" "strings" "google.golang.org/protobuf/encoding/protojson" @@ -216,7 +215,6 @@ func (projector *HistoryProjector) ProjectPromptReplay(conversation *Conversatio if ok { replayMessage.Name = toolName replayMessage.Content = limitProjectedToolResultReplay(toolName, replayMessage.Content, payload.ResultText, true, historicalToolResult) - attachReadImageContentParts(&replayMessage, toolCall) messages = append(messages, toModelMessage(replayMessage)) continue } @@ -255,7 +253,6 @@ func (projector *HistoryProjector) ProjectPromptReplay(conversation *Conversatio if strings.TrimSpace(replayMessages[index].Role) == "tool" { toolName := firstNonEmpty(strings.TrimSpace(replayMessages[index].Name), strings.TrimSpace(payload.ToolName)) replayMessages[index].Content = limitProjectedToolResultReplay(toolName, replayMessages[index].Content, payload.ResultText, true, historicalToolResult) - attachReadImageContentParts(&replayMessages[index], toolCall) } messages = append(messages, toModelMessage(replayMessages[index])) } @@ -308,58 +305,6 @@ func (projector *HistoryProjector) ProjectPromptReplay(conversation *Conversatio return normalizeReplayMessageSequence(messages), nil } -func attachReadImageContentParts(message *promptengine.Message, toolCall *agentv1.ToolCall) { - if message == nil || toolCall == nil || strings.TrimSpace(message.Role) != "tool" { - return - } - readToolCall := toolCall.GetReadToolCall() - if readToolCall == nil { - return - } - success := readToolCall.GetResult().GetSuccess() - if success == nil { - return - } - data := success.GetData() - mimeType := supportedReadImageMIMEType(data) - if mimeType == "" { - return - } - path := firstNonEmpty(strings.TrimSpace(success.GetPath()), strings.TrimSpace(readToolCall.GetArgs().GetPath())) - summary := fmt.Sprintf("read image path=%q mime=%s bytes=%d", path, mimeType, len(data)) - if fileSize := success.GetFileSize(); fileSize > 0 && uint64(fileSize) != uint64(len(data)) { - summary += fmt.Sprintf(" file_size=%d", fileSize) - } - if success.GetExceededLimit() { - summary += " truncated=true" - } - message.Content = summary - message.ContentParts = []promptengine.ContentPart{ - {Type: "text", Text: summary}, - { - Type: "image", - Image: &promptengine.ImageContent{ - MIMEType: mimeType, - Path: path, - Data: append([]byte(nil), data...), - }, - }, - } -} - -func supportedReadImageMIMEType(data []byte) string { - if len(data) == 0 { - return "" - } - mimeType := strings.ToLower(strings.TrimSpace(http.DetectContentType(data))) - switch mimeType { - case "image/png", "image/jpeg", "image/gif", "image/webp": - return mimeType - default: - return "" - } -} - func compactedPromptProjectionEntries(entries []HistoryEntry) []HistoryEntry { if len(entries) == 0 { return nil diff --git a/internal/backend/forwarder/projector_read_image_test.go b/internal/backend/forwarder/projector_read_image_test.go index 5f9d7e4..da8b11e 100644 --- a/internal/backend/forwarder/projector_read_image_test.go +++ b/internal/backend/forwarder/projector_read_image_test.go @@ -2,98 +2,179 @@ package forwarder import ( "bytes" - "fmt" + "crypto/sha256" + "encoding/base64" + "encoding/json" + "image" + "image/color" + "image/png" + "reflect" "strings" "testing" "google.golang.org/protobuf/encoding/protojson" "cursor/gen/agentv1" + modeladapter "cursor/internal/backend/agent/model" ) -func TestProjectPromptReplayAttachesReadImageToToolMessage(t *testing.T) { - testCases := []struct { - name string - imageData []byte - fileSize uint32 - wantSummary string - }{ - { - name: "small image omits result json base64", - imageData: append([]byte("\x89PNG\r\n\x1a\n"), bytes.Repeat([]byte{0}, 64)...), - fileSize: 391998, - wantSummary: `read image path="diagram.png" mime=image/png bytes=72 file_size=391998`, - }, - { - name: "large image omits replay truncation notice", - imageData: append([]byte("\x89PNG\r\n\x1a\n"), bytes.Repeat([]byte{0}, projectedReadReplayLimit)...), - fileSize: uint32(projectedReadReplayLimit + 8), - wantSummary: fmt.Sprintf(`read image path="diagram.png" mime=image/png bytes=%d`, projectedReadReplayLimit+8), - }, +func TestReadImageProjectionIsProviderOnlyAndIdempotent(t *testing.T) { + imageData := validForwarderTestPNG(t) + blobID := sha256.Sum256(imageData) + store := NewContentBlobStore(t.TempDir()) + if err := store.Put(blobID[:], imageData); err != nil { + t.Fatalf("Put() error = %v", err) + } + conversation := readImageConversation(t, blobID[:], len(imageData)) + + projector := NewHistoryProjector() + canonical, err := projector.ProjectPromptReplay(conversation) + if err != nil { + t.Fatalf("ProjectPromptReplay() error = %v", err) + } + if len(canonical) != 2 { + t.Fatalf("canonical message count = %d, want 2", len(canonical)) + } + if len(canonical[1].ContentParts) != 0 { + t.Fatalf("canonical replay contains image parts: %#v", canonical[1].ContentParts) } - for _, testCase := range testCases { - t.Run(testCase.name, func(t *testing.T) { - toolCall := &agentv1.ToolCall{ - Tool: &agentv1.ToolCall_ReadToolCall{ - ReadToolCall: &agentv1.ReadToolCall{ - Args: &agentv1.ReadToolArgs{Path: "diagram.png"}, - Result: &agentv1.ReadToolResult{ - Result: &agentv1.ReadToolResult_Success{ - Success: &agentv1.ReadToolSuccess{ - FileSize: testCase.fileSize, - Path: "diagram.png", - Output: &agentv1.ReadToolSuccess_Data{Data: testCase.imageData}, - }, - }, + first, err := enrichProviderReadImages(canonical, conversation, store) + if err != nil { + t.Fatalf("first enrichment error = %v", err) + } + second, err := enrichProviderReadImages(canonical, conversation, store) + if err != nil { + t.Fatalf("second enrichment error = %v", err) + } + if !reflect.DeepEqual(first, second) { + t.Fatalf("provider enrichment is not idempotent\nfirst=%#v\nsecond=%#v", first, second) + } + reenriched, err := enrichProviderReadImages(first, conversation, store) + if err != nil { + t.Fatalf("re-enrichment error = %v", err) + } + if !reflect.DeepEqual(first, reenriched) { + t.Fatalf("provider enrichment changed an already enriched projection\nfirst=%#v\nreenriched=%#v", first, reenriched) + } + assertProviderReadImageMessage(t, first[1], imageData) + first[1].ContentParts[1].Image.Data[0] ^= 0xff + if bytes.Equal(first[1].ContentParts[1].Image.Data, second[1].ContentParts[1].Image.Data) { + t.Fatal("separate enrichments share mutable image bytes") + } + + contextJSON, err := json.Marshal(conversation) + if err != nil { + t.Fatalf("marshal conversation: %v", err) + } + if bytes.Contains(contextJSON, imageData) || strings.Contains(string(contextJSON), base64.StdEncoding.EncodeToString(imageData)) { + t.Fatal("canonical conversation contains raw image bytes") + } + checkpoint, err := projector.ProjectCheckpointProjection(conversation) + if err != nil { + t.Fatalf("ProjectCheckpointProjection() error = %v", err) + } + checkpointJSON, err := json.Marshal(checkpoint) + if err != nil { + t.Fatalf("marshal checkpoint: %v", err) + } + if bytes.Contains(checkpointJSON, imageData) || strings.Contains(string(checkpointJSON), base64.StdEncoding.EncodeToString(imageData)) { + t.Fatal("checkpoint contains raw image bytes") + } +} + +func TestProviderReadImageEnrichmentLeavesTextReadUnchanged(t *testing.T) { + toolCall := &agentv1.ToolCall{ + Tool: &agentv1.ToolCall_ReadToolCall{ + ReadToolCall: &agentv1.ReadToolCall{ + Args: &agentv1.ReadToolArgs{Path: "notes.txt"}, + Result: &agentv1.ReadToolResult{ + Result: &agentv1.ReadToolResult_Success{ + Success: &agentv1.ReadToolSuccess{ + Path: "notes.txt", + Output: &agentv1.ReadToolSuccess_Content{Content: "hello"}, }, }, }, - } - encodedToolCall, err := protojson.Marshal(toolCall) - if err != nil { - t.Fatalf("marshal read tool call: %v", err) - } - conversation := &ConversationFile{ - ConversationID: "conversation-1", - NextTurnSeq: 2, - Entries: []HistoryEntry{ - newToolResultEntry(1, "request-1", "call-1", "Read", `{"path":"diagram.png"}`, fmt.Sprintf("read binary bytes=%d", len(testCase.imageData)), "", encodedToolCall), - }, - } - - messages, err := NewHistoryProjector().ProjectPromptReplay(conversation) - if err != nil { - t.Fatalf("ProjectPromptReplay() error = %v", err) - } - if len(messages) != 2 { - t.Fatalf("message count = %d, want assistant tool call and tool result", len(messages)) - } - toolMessage := messages[1] - if toolMessage.Role != "tool" || toolMessage.ToolCallID != "call-1" || toolMessage.Name != "Read" { - t.Fatalf("tool message metadata = %#v", toolMessage) - } - if toolMessage.Content != testCase.wantSummary { - t.Fatalf("tool content = %q, want %q", toolMessage.Content, testCase.wantSummary) - } - for _, forbidden := range []string{`"data"`, "iVBOR", "base64", "tool result replay truncated"} { - if strings.Contains(toolMessage.Content, forbidden) { - t.Fatalf("tool content contains %q: %q", forbidden, toolMessage.Content) - } - } - if len(toolMessage.ContentParts) != 2 { - t.Fatalf("tool content parts = %d, want text and image", len(toolMessage.ContentParts)) - } - if toolMessage.ContentParts[0].Type != "text" || toolMessage.ContentParts[0].Text != testCase.wantSummary { - t.Fatalf("tool text part = %#v", toolMessage.ContentParts[0]) - } - image := toolMessage.ContentParts[1].Image - if toolMessage.ContentParts[1].Type != "image" || image == nil { - t.Fatalf("tool image part = %#v", toolMessage.ContentParts[1]) - } - if image.MIMEType != "image/png" || image.Path != "diagram.png" || !bytes.Equal(image.Data, testCase.imageData) { - t.Fatalf("tool image = %#v", image) - } - }) + }, + }, + } + encoded, err := protojson.Marshal(toolCall) + if err != nil { + t.Fatalf("marshal tool call: %v", err) + } + conversation := &ConversationFile{Entries: []HistoryEntry{ + newToolResultEntry(1, "request-1", "call-1", "Read", `{"path":"notes.txt"}`, "hello", "", encoded), + }} + messages := []modeladapter.Message{{Role: "tool", ToolCallID: "call-1", Name: "Read", Content: "hello"}} + got, err := enrichProviderReadImages(messages, conversation, NewContentBlobStore(t.TempDir())) + if err != nil { + t.Fatalf("enrichProviderReadImages() error = %v", err) + } + if !reflect.DeepEqual(got, messages) { + t.Fatalf("text read changed: got=%#v want=%#v", got, messages) } } + +func readImageConversation(t *testing.T, blobID []byte, fileSize int) *ConversationFile { + t.Helper() + toolCall := &agentv1.ToolCall{ + Tool: &agentv1.ToolCall_ReadToolCall{ + ReadToolCall: &agentv1.ReadToolCall{ + Args: &agentv1.ReadToolArgs{Path: "diagram.png"}, + Result: &agentv1.ReadToolResult{ + Result: &agentv1.ReadToolResult_Success{ + Success: &agentv1.ReadToolSuccess{ + FileSize: uint32(fileSize), + Path: "diagram.png", + Output: &agentv1.ReadToolSuccess_DataBlobId{DataBlobId: append([]byte(nil), blobID...)}, + }, + }, + }, + }, + }, + } + encoded, err := protojson.Marshal(toolCall) + if err != nil { + t.Fatalf("marshal tool call: %v", err) + } + return &ConversationFile{ + ConversationID: "conversation-1", + Mode: "agent", + NextTurnSeq: 2, + Entries: []HistoryEntry{ + newToolResultEntry(1, "request-1", "call-1", "Read", `{"path":"diagram.png"}`, "read binary bytes", "", encoded), + }, + } +} + +func assertProviderReadImageMessage(t *testing.T, message modeladapter.Message, imageData []byte) { + t.Helper() + if message.Role != "tool" || message.ToolCallID != "call-1" || message.Name != "Read" { + t.Fatalf("tool message metadata = %#v", message) + } + if len(message.ContentParts) != 2 { + t.Fatalf("content part count = %d, want text and image", len(message.ContentParts)) + } + if message.ContentParts[0].Type != "text" || message.ContentParts[0].Text != message.Content { + t.Fatalf("text content part = %#v", message.ContentParts[0]) + } + imagePart := message.ContentParts[1] + if imagePart.Type != "image" || imagePart.Image == nil { + t.Fatalf("image content part = %#v", imagePart) + } + if imagePart.Image.MIMEType != "image/png" || imagePart.Image.Path != "diagram.png" || !bytes.Equal(imagePart.Image.Data, imageData) { + t.Fatalf("image content = %#v", imagePart.Image) + } +} + +func validForwarderTestPNG(t *testing.T) []byte { + t.Helper() + value := image.NewRGBA(image.Rect(0, 0, 2, 2)) + value.Set(0, 0, color.RGBA{R: 0x44, G: 0x88, B: 0xcc, A: 0xff}) + var encoded bytes.Buffer + if err := png.Encode(&encoded, value); err != nil { + t.Fatalf("encode test png: %v", err) + } + return encoded.Bytes() +} diff --git a/internal/backend/forwarder/provider_read_images.go b/internal/backend/forwarder/provider_read_images.go new file mode 100644 index 0000000..0884e22 --- /dev/null +++ b/internal/backend/forwarder/provider_read_images.go @@ -0,0 +1,174 @@ +// provider_read_images.go 负责在 provider 请求边界按 blob 引用补全 Read 图片。 +package forwarder + +import ( + "bytes" + "encoding/json" + "fmt" + "image" + _ "image/gif" + _ "image/jpeg" + _ "image/png" + "net/http" + "strings" + + "google.golang.org/protobuf/encoding/protojson" + + "cursor/gen/agentv1" + modeladapter "cursor/internal/backend/agent/model" +) + +type contentBlobReader interface { + Get(id []byte) ([]byte, error) +} + +type providerReadImageReference struct { + blobID []byte + path string + fileSize uint32 +} + +// enrichProviderReadImages 只为本次 provider 请求加载图片，不修改 canonical history 投影。 +func enrichProviderReadImages(messages []modeladapter.Message, conversation *ConversationFile, blobs contentBlobReader) ([]modeladapter.Message, error) { + cloned := cloneProviderEnrichmentMessages(messages) + references, err := collectProviderReadImageReferences(conversation) + if err != nil { + return nil, err + } + if len(references) == 0 { + return cloned, nil + } + for index := range cloned { + message := &cloned[index] + if strings.TrimSpace(message.Role) != "tool" { + continue + } + reference, ok := references[strings.TrimSpace(message.ToolCallID)] + if !ok { + continue + } + if blobs == nil { + return nil, fmt.Errorf("provider read image blob store is not initialized") + } + data, err := blobs.Get(reference.blobID) + if err != nil { + return nil, fmt.Errorf("load read image blob for tool call %s: %w", message.ToolCallID, err) + } + mimeType := validatedProviderReadImageMIMEType(data) + if mimeType == "" { + return nil, fmt.Errorf("read image blob for tool call %s is not a supported image", message.ToolCallID) + } + summary := "Read image file: " + reference.path + message.Content = summary + message.ContentParts = []modeladapter.ContentPart{ + {Type: "text", Text: summary}, + { + Type: "image", + Image: &modeladapter.ImageContent{ + MIMEType: mimeType, + Path: reference.path, + Data: append([]byte(nil), data...), + }, + }, + } + } + return cloned, nil +} + +func collectProviderReadImageReferences(conversation *ConversationFile) (map[string]providerReadImageReference, error) { + references := make(map[string]providerReadImageReference) + if conversation == nil { + return references, nil + } + for _, entry := range conversation.Entries { + if strings.TrimSpace(entry.Kind) != "tool_result" { + continue + } + var payload toolResultEntryPayload + if err := json.Unmarshal(entry.Payload, &payload); err != nil { + return nil, fmt.Errorf("decode read image tool result entry: %w", err) + } + if len(payload.ToolCall) == 0 { + continue + } + toolCall := &agentv1.ToolCall{} + if err := protojson.Unmarshal(payload.ToolCall, toolCall); err != nil { + return nil, fmt.Errorf("decode read image tool call: %w", err) + } + readToolCall := toolCall.GetReadToolCall() + if readToolCall == nil || readToolCall.GetResult().GetSuccess() == nil { + continue + } + success := readToolCall.GetResult().GetSuccess() + blobID := success.GetDataBlobId() + if len(blobID) == 0 { + continue + } + toolCallID := strings.TrimSpace(firstNonEmpty(payload.ToolCallID, entry.ToolCallID)) + if toolCallID == "" { + continue + } + reference := providerReadImageReference{ + blobID: append([]byte(nil), blobID...), + path: firstNonEmpty(strings.TrimSpace(success.GetPath()), strings.TrimSpace(readToolCall.GetArgs().GetPath())), + fileSize: success.GetFileSize(), + } + if existing, ok := references[toolCallID]; ok { + if !bytes.Equal(existing.blobID, reference.blobID) || existing.path != reference.path || existing.fileSize != reference.fileSize { + return nil, fmt.Errorf("conflicting read image references for tool call %s", toolCallID) + } + continue + } + references[toolCallID] = reference + } + return references, nil +} + +func cloneProviderEnrichmentMessages(messages []modeladapter.Message) []modeladapter.Message { + if len(messages) == 0 { + return nil + } + cloned := make([]modeladapter.Message, 0, len(messages)) + for _, message := range messages { + item := cloneReplayModelMessage(message) + if len(message.ContentParts) > 0 { + item.ContentParts = make([]modeladapter.ContentPart, len(message.ContentParts)) + for index, part := range message.ContentParts { + item.ContentParts[index] = part + if part.Image != nil { + imageCopy := *part.Image + imageCopy.Data = append([]byte(nil), part.Image.Data...) + item.ContentParts[index].Image = &imageCopy + } + } + } + cloned = append(cloned, item) + } + return cloned +} + +func validatedProviderReadImageMIMEType(data []byte) string { + if len(data) == 0 { + return "" + } + detected := strings.ToLower(strings.TrimSpace(http.DetectContentType(data))) + configuration, format, err := image.DecodeConfig(bytes.NewReader(data)) + if err != nil || configuration.Width <= 0 || configuration.Height <= 0 { + return "" + } + switch strings.ToLower(strings.TrimSpace(format)) { + case "png": + if detected == "image/png" { + return detected + } + case "jpeg": + if detected == "image/jpeg" { + return detected + } + case "gif": + if detected == "image/gif" { + return detected + } + } + return "" +} diff --git a/internal/backend/forwarder/service.go b/internal/backend/forwarder/service.go index 4cf5c5d..afa5b77 100644 --- a/internal/backend/forwarder/service.go +++ b/internal/backend/forwarder/service.go @@ -246,6 +246,7 @@ func subagentModelOverrideSummaries(overrides map[string]runtimecore.SubagentMod type Service struct { store *ConversationFileStore + contentBlobs *ContentBlobStore usageStore *UsageFileStore codebaseIndexStore *CodebaseIndexStore docsIndexStore *DocsIndexStore @@ -272,6 +273,7 @@ type agentModelMemory interface { func NewService(historyRoot string, resolver modeladapter.ChannelResolver) *Service { projector := NewHistoryProjector() store := NewConversationFileStore(historyRoot) + contentBlobs := NewContentBlobStore(historyRoot) broker := NewStreamBroker() rules := NewUserRuleStore(appdata.RulesRootPath()) var modelMemory agentModelMemory @@ -285,12 +287,13 @@ func NewService(historyRoot string, resolver modeladapter.ChannelResolver) *Serv debug := newDebugRecorder(historyRoot, broker, debugConfig) service := &Service{ store: store, + contentBlobs: contentBlobs, usageStore: NewUsageFileStore(historyRoot), codebaseIndexStore: NewCodebaseIndexStore(appdata.CodebaseIndexRootPath()), docsIndexStore: NewDocsIndexStore(appdata.DocsIndexRootPath()), rules: rules, projector: projector, - compiler: NewPromptCompiler(projector, NewToolCatalog(), NewReminderInjector(), rules), + compiler: NewPromptCompiler(projector, NewToolCatalog(), NewReminderInjector(), rules, contentBlobs), provider: NewProviderGateway(resolver), resolver: resolver, modelMemory: modelMemory, @@ -315,6 +318,7 @@ func newServiceWithDependencies(store *ConversationFileStore, projector *History debug := newDebugRecorder(historyRoot, broker, nil) return &Service{ store: store, + contentBlobs: NewContentBlobStore(historyRoot), rules: NewUserRuleStore(appdata.RulesRootPath()), projector: projector, compiler: compiler, @@ -914,6 +918,9 @@ func (service *Service) handleExecResult(intent InboundIntent) error { if !result.IsTerminal { return nil } + if err := service.persistExecContentBlobs(result.ContentBlobs); err != nil { + return err + } markExecCompleted(stream, pending) backgroundShellToolCallID := "" if strings.TrimSpace(pending.ExecKind) == "shell" && shellToolCallIsBackgrounded(result.ToolCall) { @@ -952,6 +959,21 @@ func (service *Service) handleExecResult(intent InboundIntent) error { return service.reconcileStream(stream) } +func (service *Service) persistExecContentBlobs(blobs []execbridge.ContentBlob) error { + if len(blobs) == 0 { + return nil + } + if service == nil || service.contentBlobs == nil { + return fmt.Errorf("content blob store is not initialized") + } + for _, blob := range blobs { + if err := service.contentBlobs.Put(blob.ID, blob.Data); err != nil { + return fmt.Errorf("persist exec content blob: %w", err) + } + } + return nil +} + // handleExecControl 处理执行桥控制面结果，例如 stream_close 或 throw。 func (service *Service) handleExecControl(intent InboundIntent) error { stream, ok := service.broker.Get(intent.RequestID)