From 836220363154452217f7555c1c02c74e24b8384b Mon Sep 17 00:00:00 2001 From: Anton Bogdanovich <27antonb@gmail.com> Date: Fri, 8 May 2026 13:50:14 -0700 Subject: [PATCH] fix(agent): transcribe queued voice follow-ups --- pkg/agent/agent.go | 2 + pkg/agent/agent_message.go | 63 +++++++++++++++--- pkg/agent/steering_test.go | 127 +++++++++++++++++++++++++++++++++++++ 3 files changed, 182 insertions(+), 10 deletions(-) diff --git a/pkg/agent/agent.go b/pkg/agent/agent.go index 97ee4fe7d..bc6d2b39b 100644 --- a/pkg/agent/agent.go +++ b/pkg/agent/agent.go @@ -182,6 +182,8 @@ func (al *AgentLoop) Run(ctx context.Context) error { continue } + msg = al.prepareInboundMessageForAgent(ctx, msg) + // Another turn is already active (or reserved) for this session — enqueue if err := al.enqueueSteeringMessage(sessionKey, agentID, providers.Message{ Role: "user", diff --git a/pkg/agent/agent_message.go b/pkg/agent/agent_message.go index 96b0b0817..8fba50d71 100644 --- a/pkg/agent/agent_message.go +++ b/pkg/agent/agent_message.go @@ -65,6 +65,40 @@ func (al *AgentLoop) ProcessDirectWithChannel( return al.processMessage(ctx, msg) } +func (al *AgentLoop) processScheduledMessage(ctx context.Context, msg bus.InboundMessage) (string, error) { + msg = al.prepareInboundMessageForAgent(ctx, msg) + route, agent, routeErr := al.resolveMessageRoute(msg) + if routeErr != nil { + return "", routeErr + } + allocation := al.allocateRouteSession(route, msg) + sessionKey := resolveScopeKey(allocation.SessionKey, msg.SessionKey) + + if tool, ok := agent.Tools.Get("message"); ok { + if resetter, ok := tool.(interface{ ResetSentInRound(sessionKey string) }); ok { + resetter.ResetSentInRound(sessionKey) + } + } + + return al.runAgentLoop(ctx, agent, processOptions{ + Dispatch: DispatchRequest{ + SessionKey: sessionKey, + SessionAliases: buildSessionAliases(sessionKey, append(allocation.SessionAliases, msg.SessionKey)...), + InboundContext: cloneInboundContext(&msg.Context), + RouteResult: cloneResolvedRoute(&route), + SessionScope: session.CloneScope(&allocation.Scope), + UserMessage: msg.Content, + Media: append([]string(nil), msg.Media...), + }, + SenderID: msg.SenderID, + SenderDisplayName: msg.Sender.DisplayName, + DefaultResponse: defaultResponse, + EnableSummary: false, + SendResponse: false, + SuppressToolFeedback: true, + NoHistory: true, + }) +} func (al *AgentLoop) ProcessHeartbeat( ctx context.Context, content, channel, chatID string, @@ -102,9 +136,27 @@ func (al *AgentLoop) ProcessHeartbeat( }) } -func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) (string, error) { +func (al *AgentLoop) prepareInboundMessageForAgent( + ctx context.Context, + msg bus.InboundMessage, +) bus.InboundMessage { msg = bus.NormalizeInboundMessage(msg) + var hadAudio bool + msg, hadAudio = al.transcribeAudioInMessage(ctx, msg) + + // For audio messages the placeholder was deferred by the channel. + // Now that transcription (and optional feedback) is done, send it. + if hadAudio && al.channelManager != nil { + al.channelManager.SendPlaceholder(ctx, msg.Channel, msg.ChatID) + } + + return msg +} + +func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) (string, error) { + msg = al.prepareInboundMessageForAgent(ctx, msg) + // Add message preview to log (show full content for error messages) var logContent string if strings.Contains(msg.Content, "Error:") || strings.Contains(msg.Content, "error") { @@ -123,15 +175,6 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) }, ) - var hadAudio bool - msg, hadAudio = al.transcribeAudioInMessage(ctx, msg) - - // For audio messages the placeholder was deferred by the channel. - // Now that transcription (and optional feedback) is done, send it. - if hadAudio && al.channelManager != nil { - al.channelManager.SendPlaceholder(ctx, msg.Channel, msg.ChatID) - } - // Route system messages to processSystemMessage if msg.Channel == "system" { return al.processSystemMessage(ctx, msg) diff --git a/pkg/agent/steering_test.go b/pkg/agent/steering_test.go index 813013649..23d34840e 100644 --- a/pkg/agent/steering_test.go +++ b/pkg/agent/steering_test.go @@ -12,6 +12,7 @@ import ( "testing" "time" + "github.com/sipeed/picoclaw/pkg/audio/asr" "github.com/sipeed/picoclaw/pkg/bus" "github.com/sipeed/picoclaw/pkg/config" runtimeevents "github.com/sipeed/picoclaw/pkg/events" @@ -477,6 +478,16 @@ func (p *lateSteeringProvider) GetDefaultModel() string { return "late-steering-mock" } +type fixedTranscriber struct { + text string +} + +func (f *fixedTranscriber) Name() string { return "fixed" } + +func (f *fixedTranscriber) Transcribe(ctx context.Context, audioFilePath string) (*asr.TranscriptionResponse, error) { + return &asr.TranscriptionResponse{Text: f.text}, nil +} + type blockingDirectProvider struct { mu sync.Mutex calls int @@ -840,6 +851,122 @@ func TestAgentLoop_Run_AutoContinuesLateSteeringMessage(t *testing.T) { } } +func TestAgentLoop_Run_QueuedVoiceMessageIsTranscribedBeforeSteering(t *testing.T) { + tmpDir := t.TempDir() + cfg := &config.Config{ + Agents: config.AgentsConfig{ + Defaults: config.AgentDefaults{ + Workspace: tmpDir, + ModelName: "test-model", + MaxTokens: 4096, + MaxToolIterations: 10, + }, + }, + } + + msgBus := bus.NewMessageBus() + provider := &lateSteeringProvider{ + firstCallStarted: make(chan struct{}), + releaseFirstCall: make(chan struct{}), + } + al := NewAgentLoop(cfg, msgBus, provider) + + store := media.NewFileMediaStore() + audioPath := filepath.Join(tmpDir, "voice.ogg") + if err := os.WriteFile(audioPath, []byte("fake audio"), 0o644); err != nil { + t.Fatalf("write audio fixture: %v", err) + } + ref, err := store.Store(audioPath, media.MediaMeta{ + Filename: "voice.ogg", + ContentType: "audio/ogg", + CleanupPolicy: media.CleanupPolicyForgetOnly, + }, "scope-voice") + if err != nil { + t.Fatalf("store audio fixture: %v", err) + } + al.SetMediaStore(store) + al.SetTranscriber(&fixedTranscriber{text: "and also two pieces of bread"}) + + runCtx, cancelRun := context.WithCancel(context.Background()) + defer cancelRun() + + runErrCh := make(chan error, 1) + go func() { + runErrCh <- al.Run(runCtx) + }() + + first := bus.InboundMessage{ + Context: bus.InboundContext{ + Channel: "test", + ChatID: "chat1", + ChatType: "direct", + SenderID: "user1", + }, + Content: "first meal", + } + late := bus.InboundMessage{ + Context: bus.InboundContext{ + Channel: "test", + ChatID: "chat1", + ChatType: "direct", + SenderID: "user1", + }, + Content: "[voice]", + Media: []string{ref}, + } + + pubCtx, pubCancel := context.WithTimeout(context.Background(), 2*time.Second) + defer pubCancel() + if err := msgBus.PublishInbound(pubCtx, first); err != nil { + t.Fatalf("publish first inbound: %v", err) + } + + select { + case <-provider.firstCallStarted: + case <-time.After(2 * time.Second): + t.Fatal("timeout waiting for first provider call to start") + } + + if err := msgBus.PublishInbound(pubCtx, late); err != nil { + t.Fatalf("publish late voice inbound: %v", err) + } + + close(provider.releaseFirstCall) + + subCtx, subCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer subCancel() + select { + case <-msgBus.OutboundChan(): + case <-subCtx.Done(): + t.Fatal("expected outbound response") + } + + cancelRun() + select { + case err := <-runErrCh: + if err != nil { + t.Fatalf("Run returned error: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("timeout waiting for Run to stop") + } + + provider.mu.Lock() + secondMessages := append([]providers.Message(nil), provider.secondCallMessages...) + provider.mu.Unlock() + + foundTranscribedVoice := false + for _, msg := range secondMessages { + if msg.Role == "user" && strings.Contains(msg.Content, "[voice: and also two pieces of bread]") { + foundTranscribedVoice = true + break + } + } + if !foundTranscribedVoice { + t.Fatalf("expected queued voice message to be transcribed before steering injection, got %#v", secondMessages) + } +} + func TestAgentLoop_Run_PendingStopStillContinuesQueuedFollowUp(t *testing.T) { tmpDir, err := os.MkdirTemp("", "agent-test-*") if err != nil {