fix(backend): persist attachments on queued message when dequeued (#659)
This commit is contained in:
@@ -220,6 +220,15 @@ func (s *Service) executeQueuedMessage(callerSessionID string, queuedMsg *messag
|
||||
return
|
||||
}
|
||||
|
||||
attachments := make([]v1.MessageAttachment, len(queuedMsg.Attachments))
|
||||
for i, att := range queuedMsg.Attachments {
|
||||
attachments[i] = v1.MessageAttachment{
|
||||
Type: att.Type,
|
||||
Data: att.Data,
|
||||
MimeType: att.MimeType,
|
||||
}
|
||||
}
|
||||
|
||||
// Create user message for the queued message (so it appears in chat history)
|
||||
if s.messageCreator != nil {
|
||||
turnID := s.getActiveTurnID(queuedMsg.SessionID)
|
||||
@@ -229,9 +238,9 @@ func (s *Service) executeQueuedMessage(callerSessionID string, queuedMsg *messag
|
||||
turnID = s.getActiveTurnID(queuedMsg.SessionID)
|
||||
}
|
||||
|
||||
// Note: Attachments will be sent to the agent via PromptTask but not stored in the message
|
||||
// This matches the behavior of direct prompts
|
||||
meta := NewUserMessageMeta().WithPlanMode(queuedMsg.PlanMode)
|
||||
meta := NewUserMessageMeta().
|
||||
WithPlanMode(queuedMsg.PlanMode).
|
||||
WithAttachments(attachments)
|
||||
err := s.messageCreator.CreateUserMessage(promptCtx, queuedMsg.TaskID, queuedMsg.Content, queuedMsg.SessionID, turnID, meta.ToMap())
|
||||
if err != nil {
|
||||
s.logger.Error("failed to create user message for queued message",
|
||||
@@ -248,16 +257,6 @@ func (s *Service) executeQueuedMessage(callerSessionID string, queuedMsg *messag
|
||||
s.processOnTurnStartViaEngine(promptCtx, queuedMsg.TaskID, session)
|
||||
}
|
||||
|
||||
// Convert queue attachments to v1 attachments
|
||||
attachments := make([]v1.MessageAttachment, len(queuedMsg.Attachments))
|
||||
for i, att := range queuedMsg.Attachments {
|
||||
attachments[i] = v1.MessageAttachment{
|
||||
Type: att.Type,
|
||||
Data: att.Data,
|
||||
MimeType: att.MimeType,
|
||||
}
|
||||
}
|
||||
|
||||
_, err := s.PromptTask(promptCtx, queuedMsg.TaskID, queuedMsg.SessionID,
|
||||
queuedMsg.Content, queuedMsg.Model, queuedMsg.PlanMode, attachments)
|
||||
if err != nil {
|
||||
|
||||
@@ -2,11 +2,13 @@ package orchestrator
|
||||
|
||||
import (
|
||||
"context"
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
"github.com/kandev/kandev/internal/orchestrator/executor"
|
||||
"github.com/kandev/kandev/internal/orchestrator/messagequeue"
|
||||
"github.com/kandev/kandev/internal/task/models"
|
||||
v1 "github.com/kandev/kandev/pkg/api/v1"
|
||||
)
|
||||
|
||||
func TestExecuteQueuedMessage_RequeuesWhenResetInProgress(t *testing.T) {
|
||||
@@ -47,3 +49,63 @@ func TestExecuteQueuedMessage_RequeuesWhenResetInProgress(t *testing.T) {
|
||||
t.Fatalf("expected queued content to be preserved, got %q", status.Message.Content)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecuteQueuedMessage_StoresAttachmentsInUserMessageMetadata(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
repo := setupTestRepo(t)
|
||||
seedSession(t, repo, "t1", "s1", "step1")
|
||||
|
||||
session, err := repo.GetTaskSession(ctx, "s1")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to get session: %v", err)
|
||||
}
|
||||
session.State = models.TaskSessionStateWaitingForInput
|
||||
session.AgentExecutionID = "exec-1"
|
||||
if err := repo.UpdateTaskSession(ctx, session); err != nil {
|
||||
t.Fatalf("failed to update session: %v", err)
|
||||
}
|
||||
|
||||
taskRepo := newMockTaskRepo()
|
||||
agentMgr := &mockAgentManager{isAgentRunning: true}
|
||||
svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, agentMgr)
|
||||
svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{})
|
||||
|
||||
mc := &mockMessageCreator{}
|
||||
svc.messageCreator = mc
|
||||
|
||||
queuedAtts := []messagequeue.MessageAttachment{
|
||||
{Type: "image", Data: "base64payload", MimeType: "image/png"},
|
||||
}
|
||||
queuedMsg := &messagequeue.QueuedMessage{
|
||||
ID: "q1",
|
||||
SessionID: "s1",
|
||||
TaskID: "t1",
|
||||
Content: "look at this screenshot",
|
||||
Attachments: queuedAtts,
|
||||
QueuedBy: "test",
|
||||
}
|
||||
|
||||
svc.executeQueuedMessage("s1", queuedMsg)
|
||||
|
||||
if len(mc.userMessages) != 1 {
|
||||
t.Fatalf("expected 1 user message recorded, got %d", len(mc.userMessages))
|
||||
}
|
||||
meta := mc.userMessages[0].metadata
|
||||
if meta == nil {
|
||||
t.Fatalf("expected metadata on user message, got nil")
|
||||
}
|
||||
raw, ok := meta["attachments"]
|
||||
if !ok {
|
||||
t.Fatalf("expected metadata to contain 'attachments' key, got %v", meta)
|
||||
}
|
||||
got, ok := raw.([]v1.MessageAttachment)
|
||||
if !ok {
|
||||
t.Fatalf("expected attachments to be []v1.MessageAttachment, got %T", raw)
|
||||
}
|
||||
want := []v1.MessageAttachment{
|
||||
{Type: "image", Data: "base64payload", MimeType: "image/png"},
|
||||
}
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("attachments mismatch\n got: %+v\nwant: %+v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user