Merge remote-tracking branch 'origin/main' into feat/telegram-use-md2
This commit is contained in:
Aleksandr Bortnikov
commit
7fc601b351
23 changed files with 687 additions and 46 deletions
|
|
@ -1,23 +1,42 @@
|
|||
package gateway
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"github.com/spf13/cobra"
|
||||
|
||||
"github.com/sipeed/picoclaw/pkg/logger"
|
||||
"github.com/sipeed/picoclaw/pkg/utils"
|
||||
)
|
||||
|
||||
func NewGatewayCommand() *cobra.Command {
|
||||
var debug bool
|
||||
var noTruncate bool
|
||||
|
||||
cmd := &cobra.Command{
|
||||
Use: "gateway",
|
||||
Aliases: []string{"g"},
|
||||
Short: "Start picoclaw gateway",
|
||||
Args: cobra.NoArgs,
|
||||
PreRunE: func(_ *cobra.Command, _ []string) error {
|
||||
if noTruncate && !debug {
|
||||
return fmt.Errorf("the --no-truncate option can only be used in conjunction with --debug (-d)")
|
||||
}
|
||||
|
||||
if noTruncate {
|
||||
utils.SetDisableTruncation(true)
|
||||
logger.Info("String truncation is globally disabled via 'no-truncate' flag")
|
||||
}
|
||||
|
||||
return nil
|
||||
},
|
||||
RunE: func(_ *cobra.Command, _ []string) error {
|
||||
return gatewayCmd(debug)
|
||||
},
|
||||
}
|
||||
|
||||
cmd.Flags().BoolVarP(&debug, "debug", "d", false, "Enable debug logging")
|
||||
cmd.Flags().BoolVarP(&noTruncate, "no-truncate", "T", false, "Disable string truncation in debug logs")
|
||||
|
||||
return cmd
|
||||
}
|
||||
|
|
|
|||
|
|
@ -476,6 +476,9 @@
|
|||
"enabled": false,
|
||||
"monitor_usb": true
|
||||
},
|
||||
"voice": {
|
||||
"echo_transcription": false
|
||||
},
|
||||
"gateway": {
|
||||
"host": "127.0.0.1",
|
||||
"port": 18790
|
||||
|
|
|
|||
33
docs/debug.md
Normal file
33
docs/debug.md
Normal file
|
|
@ -0,0 +1,33 @@
|
|||
# Debugging PicoClaw
|
||||
|
||||
PicoClaw performs multiple complex interactions under the hood for every single request it receives—from routing messages and evaluating complexity, to executing tools and adapting to model failures. Being able to see exactly what is happening is crucial, not just for troubleshooting potential issues, but also for truly understanding how the agent operates.
|
||||
## Starting PicoClaw in Debug Mode
|
||||
|
||||
To get detailed information about what the agent is doing (LLM requests, tool calls, message routing), you can start the PicoClaw gateway with the debug flag:
|
||||
|
||||
```bash
|
||||
picoclaw gateway --debug
|
||||
# or
|
||||
picoclaw gateway -d
|
||||
```
|
||||
|
||||
In this mode, the system will format the logs extensively and display previews of system prompts and tool execution results.
|
||||
|
||||
## Disabling Log Truncation (Full Logs)
|
||||
|
||||
By default, PicoClaw truncates very long strings (such as the *System Prompt* or large JSON output results) in the debug logs to keep the console readable.
|
||||
|
||||
If you need to inspect the complete output of a command or the exact payload sent to the LLM model, you can use the `--no-truncate` flag.
|
||||
|
||||
**Note:** This flag *only* works when combined with the `--debug` mode.
|
||||
|
||||
```bash
|
||||
picoclaw gateway --debug --no-truncate
|
||||
|
||||
```
|
||||
|
||||
When this flag is active, the global truncation function is disabled. This is extremely useful for:
|
||||
|
||||
* Verifying the exact syntax of the messages sent to the provider.
|
||||
* Reading the complete output of tools like `exec`, `web_fetch`, or `read_file`.
|
||||
* Debugging the session history saved in memory.
|
||||
|
|
@ -16,6 +16,7 @@ import (
|
|||
"github.com/sipeed/picoclaw/pkg/logger"
|
||||
"github.com/sipeed/picoclaw/pkg/providers"
|
||||
"github.com/sipeed/picoclaw/pkg/skills"
|
||||
"github.com/sipeed/picoclaw/pkg/utils"
|
||||
)
|
||||
|
||||
type ContextBuilder struct {
|
||||
|
|
@ -538,10 +539,7 @@ func (cb *ContextBuilder) BuildMessages(
|
|||
})
|
||||
|
||||
// Log preview of system prompt (avoid logging huge content)
|
||||
preview := fullSystemPrompt
|
||||
if len(preview) > 500 {
|
||||
preview = preview[:500] + "... (truncated)"
|
||||
}
|
||||
preview := utils.Truncate(fullSystemPrompt, 500)
|
||||
logger.DebugCF("agent", "System prompt preview",
|
||||
map[string]any{
|
||||
"preview": preview,
|
||||
|
|
|
|||
|
|
@ -467,9 +467,10 @@ var audioAnnotationRe = regexp.MustCompile(`\[(voice|audio)(?::[^\]]*)?\]`)
|
|||
|
||||
// transcribeAudioInMessage resolves audio media refs, transcribes them, and
|
||||
// replaces audio annotations in msg.Content with the transcribed text.
|
||||
func (al *AgentLoop) transcribeAudioInMessage(ctx context.Context, msg bus.InboundMessage) bus.InboundMessage {
|
||||
// Returns the (possibly modified) message and true if audio was transcribed.
|
||||
func (al *AgentLoop) transcribeAudioInMessage(ctx context.Context, msg bus.InboundMessage) (bus.InboundMessage, bool) {
|
||||
if al.transcriber == nil || al.mediaStore == nil || len(msg.Media) == 0 {
|
||||
return msg
|
||||
return msg, false
|
||||
}
|
||||
|
||||
// Transcribe each audio media ref in order.
|
||||
|
|
@ -493,9 +494,11 @@ func (al *AgentLoop) transcribeAudioInMessage(ctx context.Context, msg bus.Inbou
|
|||
}
|
||||
|
||||
if len(transcriptions) == 0 {
|
||||
return msg
|
||||
return msg, false
|
||||
}
|
||||
|
||||
al.sendTranscriptionFeedback(ctx, msg.Channel, msg.ChatID, msg.MessageID, transcriptions)
|
||||
|
||||
// Replace audio annotations sequentially with transcriptions.
|
||||
idx := 0
|
||||
newContent := audioAnnotationRe.ReplaceAllStringFunc(msg.Content, func(match string) string {
|
||||
|
|
@ -513,7 +516,48 @@ func (al *AgentLoop) transcribeAudioInMessage(ctx context.Context, msg bus.Inbou
|
|||
}
|
||||
|
||||
msg.Content = newContent
|
||||
return msg
|
||||
return msg, true
|
||||
}
|
||||
|
||||
// sendTranscriptionFeedback sends feedback to the user with the result of
|
||||
// audio transcription if the option is enabled. It uses Manager.SendMessage
|
||||
// which executes synchronously (rate limiting, splitting, retry) so that
|
||||
// ordering with the subsequent placeholder is guaranteed.
|
||||
func (al *AgentLoop) sendTranscriptionFeedback(
|
||||
ctx context.Context,
|
||||
channel, chatID, messageID string,
|
||||
validTexts []string,
|
||||
) {
|
||||
if !al.cfg.Voice.EchoTranscription {
|
||||
return
|
||||
}
|
||||
if al.channelManager == nil {
|
||||
return
|
||||
}
|
||||
|
||||
var nonEmpty []string
|
||||
for _, t := range validTexts {
|
||||
if t != "" {
|
||||
nonEmpty = append(nonEmpty, t)
|
||||
}
|
||||
}
|
||||
|
||||
var feedbackMsg string
|
||||
if len(nonEmpty) > 0 {
|
||||
feedbackMsg = "Transcript: " + strings.Join(nonEmpty, "\n")
|
||||
} else {
|
||||
feedbackMsg = "No voice detected in the audio"
|
||||
}
|
||||
|
||||
err := al.channelManager.SendMessage(ctx, bus.OutboundMessage{
|
||||
Channel: channel,
|
||||
ChatID: chatID,
|
||||
Content: feedbackMsg,
|
||||
ReplyToMessageID: messageID,
|
||||
})
|
||||
if err != nil {
|
||||
logger.WarnCF("voice", "Failed to send transcription feedback", map[string]any{"error": err.Error()})
|
||||
}
|
||||
}
|
||||
|
||||
// inferMediaType determines the media type ("image", "audio", "video", "file")
|
||||
|
|
@ -627,7 +671,14 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage)
|
|||
},
|
||||
)
|
||||
|
||||
msg = al.transcribeAudioInMessage(ctx, 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)
|
||||
}
|
||||
|
||||
// Route system messages to processSystemMessage
|
||||
if msg.Channel == "system" {
|
||||
|
|
|
|||
|
|
@ -33,6 +33,7 @@ type OutboundMessage struct {
|
|||
Channel string `json:"channel"`
|
||||
ChatID string `json:"chat_id"`
|
||||
Content string `json:"content"`
|
||||
ReplyToMessageID string `json:"reply_to_message_id,omitempty"`
|
||||
}
|
||||
|
||||
// MediaPart describes a single media attachment to send.
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ import (
|
|||
"crypto/rand"
|
||||
"encoding/binary"
|
||||
"encoding/hex"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
|
|
@ -32,6 +33,9 @@ func init() {
|
|||
uniqueIDPrefix = hex.EncodeToString(b[:])
|
||||
}
|
||||
|
||||
// audioAnnotationRe matches audio/voice annotations injected by channels (e.g. [voice], [audio: file.ogg]).
|
||||
var audioAnnotationRe = regexp.MustCompile(`\[(voice|audio)(?::[^\]]*)?\]`)
|
||||
|
||||
// uniqueID generates a process-unique ID using a random prefix and an atomic counter.
|
||||
// This ID is intended for internal correlation (e.g. media scope keys) and is NOT
|
||||
// cryptographically secure — it must not be used in contexts where unpredictability matters.
|
||||
|
|
@ -284,13 +288,18 @@ func (c *BaseChannel) HandleMessage(
|
|||
c.placeholderRecorder.RecordReactionUndo(c.name, chatID, undo)
|
||||
}
|
||||
}
|
||||
// Placeholder — independent pipeline
|
||||
// Placeholder — independent pipeline.
|
||||
// Skip when the message contains audio: the agent will send the
|
||||
// placeholder after transcription completes, so the user sees
|
||||
// "Thinking…" only once the voice has been processed.
|
||||
if !audioAnnotationRe.MatchString(content) {
|
||||
if pc, ok := c.owner.(PlaceholderCapable); ok {
|
||||
if phID, err := pc.SendPlaceholder(ctx, chatID); err == nil && phID != "" {
|
||||
c.placeholderRecorder.RecordPlaceholder(c.name, chatID, phID)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if err := c.bus.PublishInbound(ctx, msg); err != nil {
|
||||
logger.ErrorCF("channels", "Failed to publish inbound message", map[string]any{
|
||||
|
|
|
|||
|
|
@ -134,7 +134,7 @@ func (c *DiscordChannel) Send(ctx context.Context, msg bus.OutboundMessage) erro
|
|||
return nil
|
||||
}
|
||||
|
||||
return c.sendChunk(ctx, channelID, msg.Content)
|
||||
return c.sendChunk(ctx, channelID, msg.Content, msg.ReplyToMessageID)
|
||||
}
|
||||
|
||||
// SendMedia implements the channels.MediaSender interface.
|
||||
|
|
@ -259,14 +259,29 @@ func (c *DiscordChannel) SendPlaceholder(ctx context.Context, chatID string) (st
|
|||
return msg.ID, nil
|
||||
}
|
||||
|
||||
func (c *DiscordChannel) sendChunk(ctx context.Context, channelID, content string) error {
|
||||
func (c *DiscordChannel) sendChunk(ctx context.Context, channelID, content, replyToID string) error {
|
||||
// Use the passed ctx for timeout control
|
||||
sendCtx, cancel := context.WithTimeout(ctx, sendTimeout)
|
||||
defer cancel()
|
||||
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := c.session.ChannelMessageSend(channelID, content)
|
||||
var err error
|
||||
|
||||
// If we have an ID, we send the message as "Reply"
|
||||
if replyToID != "" {
|
||||
_, err = c.session.ChannelMessageSendComplex(channelID, &discordgo.MessageSend{
|
||||
Content: content,
|
||||
Reference: &discordgo.MessageReference{
|
||||
MessageID: replyToID,
|
||||
ChannelID: channelID,
|
||||
},
|
||||
})
|
||||
} else {
|
||||
// Otherwise, we send a normal message
|
||||
_, err = c.session.ChannelMessageSend(channelID, content)
|
||||
}
|
||||
|
||||
done <- err
|
||||
}()
|
||||
|
||||
|
|
|
|||
|
|
@ -102,6 +102,27 @@ func (m *Manager) RecordPlaceholder(channel, chatID, placeholderID string) {
|
|||
m.placeholders.Store(key, placeholderEntry{id: placeholderID, createdAt: time.Now()})
|
||||
}
|
||||
|
||||
// SendPlaceholder sends a "Thinking…" placeholder for the given channel/chatID
|
||||
// and records it for later editing. Returns true if a placeholder was sent.
|
||||
func (m *Manager) SendPlaceholder(ctx context.Context, channel, chatID string) bool {
|
||||
m.mu.RLock()
|
||||
ch, ok := m.channels[channel]
|
||||
m.mu.RUnlock()
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
pc, ok := ch.(PlaceholderCapable)
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
phID, err := pc.SendPlaceholder(ctx, chatID)
|
||||
if err != nil || phID == "" {
|
||||
return false
|
||||
}
|
||||
m.RecordPlaceholder(channel, chatID, phID)
|
||||
return true
|
||||
}
|
||||
|
||||
// RecordTypingStop registers a typing stop function for later invocation.
|
||||
// Implements PlaceholderRecorder.
|
||||
func (m *Manager) RecordTypingStop(channel, chatID string, stop func()) {
|
||||
|
|
@ -813,6 +834,39 @@ func (m *Manager) UnregisterChannel(name string) {
|
|||
delete(m.channels, name)
|
||||
}
|
||||
|
||||
// SendMessage sends an outbound message synchronously through the channel
|
||||
// worker's rate limiter and retry logic. It blocks until the message is
|
||||
// delivered (or all retries are exhausted), which preserves ordering when
|
||||
// a subsequent operation depends on the message having been sent.
|
||||
func (m *Manager) SendMessage(ctx context.Context, msg bus.OutboundMessage) error {
|
||||
m.mu.RLock()
|
||||
_, exists := m.channels[msg.Channel]
|
||||
w, wExists := m.workers[msg.Channel]
|
||||
m.mu.RUnlock()
|
||||
|
||||
if !exists {
|
||||
return fmt.Errorf("channel %s not found", msg.Channel)
|
||||
}
|
||||
if !wExists || w == nil {
|
||||
return fmt.Errorf("channel %s has no active worker", msg.Channel)
|
||||
}
|
||||
|
||||
maxLen := 0
|
||||
if mlp, ok := w.ch.(MessageLengthProvider); ok {
|
||||
maxLen = mlp.MaxMessageLength()
|
||||
}
|
||||
if maxLen > 0 && len([]rune(msg.Content)) > maxLen {
|
||||
for _, chunk := range SplitMessage(msg.Content, maxLen) {
|
||||
chunkMsg := msg
|
||||
chunkMsg.Content = chunk
|
||||
m.sendWithRetry(ctx, msg.Channel, w, chunkMsg)
|
||||
}
|
||||
} else {
|
||||
m.sendWithRetry(ctx, msg.Channel, w, msg)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manager) SendToChannel(ctx context.Context, channelName, chatID, content string) error {
|
||||
m.mu.RLock()
|
||||
_, exists := m.channels[channelName]
|
||||
|
|
|
|||
|
|
@ -18,15 +18,31 @@ import (
|
|||
type mockChannel struct {
|
||||
BaseChannel
|
||||
sendFn func(ctx context.Context, msg bus.OutboundMessage) error
|
||||
sentMessages []bus.OutboundMessage
|
||||
placeholdersSent int
|
||||
editedMessages int
|
||||
lastPlaceholderID string
|
||||
}
|
||||
|
||||
func (m *mockChannel) Send(ctx context.Context, msg bus.OutboundMessage) error {
|
||||
m.sentMessages = append(m.sentMessages, msg)
|
||||
return m.sendFn(ctx, msg)
|
||||
}
|
||||
|
||||
func (m *mockChannel) Start(ctx context.Context) error { return nil }
|
||||
func (m *mockChannel) Stop(ctx context.Context) error { return nil }
|
||||
|
||||
func (m *mockChannel) SendPlaceholder(ctx context.Context, chatID string) (string, error) {
|
||||
m.placeholdersSent++
|
||||
m.lastPlaceholderID = "mock-ph-123"
|
||||
return m.lastPlaceholderID, nil
|
||||
}
|
||||
|
||||
func (m *mockChannel) EditMessage(ctx context.Context, chatID, messageID, content string) error {
|
||||
m.editedMessages++
|
||||
return nil
|
||||
}
|
||||
|
||||
// newTestManager creates a minimal Manager suitable for unit tests.
|
||||
func newTestManager() *Manager {
|
||||
return &Manager{
|
||||
|
|
@ -860,3 +876,286 @@ func TestBuildMediaScope_WithMessageID(t *testing.T) {
|
|||
t.Fatalf("expected %s, got %s", expected, scope)
|
||||
}
|
||||
}
|
||||
|
||||
func TestManager_PlaceholderConsumedByResponse(t *testing.T) {
|
||||
mgr := &Manager{
|
||||
channels: make(map[string]Channel),
|
||||
workers: make(map[string]*channelWorker),
|
||||
placeholders: sync.Map{},
|
||||
}
|
||||
|
||||
mockCh := &mockChannel{
|
||||
sendFn: func(ctx context.Context, msg bus.OutboundMessage) error {
|
||||
return nil
|
||||
},
|
||||
}
|
||||
worker := newChannelWorker("mock", mockCh)
|
||||
mgr.channels["mock"] = mockCh
|
||||
mgr.workers["mock"] = worker
|
||||
|
||||
ctx := context.Background()
|
||||
key := "mock:chat-1"
|
||||
|
||||
// Simulate a placeholder recorded by base.go HandleMessage
|
||||
mgr.RecordPlaceholder("mock", "chat-1", "ph-123")
|
||||
|
||||
if _, ok := mgr.placeholders.Load(key); !ok {
|
||||
t.Fatal("expected placeholder to be recorded")
|
||||
}
|
||||
|
||||
// Transcription feedback arrives first — it should consume the placeholder
|
||||
// and be delivered via EditMessage, not Send.
|
||||
msgTranscript := bus.OutboundMessage{
|
||||
Channel: "mock",
|
||||
ChatID: "chat-1",
|
||||
Content: "Transcript: hello",
|
||||
}
|
||||
mgr.sendWithRetry(ctx, "mock", worker, msgTranscript)
|
||||
|
||||
if mockCh.editedMessages != 1 {
|
||||
t.Errorf("expected 1 edited message (placeholder consumed by transcript), got %d", mockCh.editedMessages)
|
||||
}
|
||||
if len(mockCh.sentMessages) != 0 {
|
||||
t.Errorf("expected 0 normal messages (transcript used edit), got %d", len(mockCh.sentMessages))
|
||||
}
|
||||
|
||||
// Placeholder should be gone now
|
||||
if _, ok := mgr.placeholders.Load(key); ok {
|
||||
t.Error("expected placeholder to be removed after being consumed")
|
||||
}
|
||||
|
||||
// Final LLM response arrives — no placeholder left, so it goes through Send
|
||||
msgFinal := bus.OutboundMessage{
|
||||
Channel: "mock",
|
||||
ChatID: "chat-1",
|
||||
Content: "Final Answer",
|
||||
}
|
||||
mgr.sendWithRetry(ctx, "mock", worker, msgFinal)
|
||||
|
||||
if len(mockCh.sentMessages) != 1 {
|
||||
t.Errorf("expected 1 normal message sent, got %d", len(mockCh.sentMessages))
|
||||
}
|
||||
}
|
||||
|
||||
func TestSendMessage_Synchronous(t *testing.T) {
|
||||
m := newTestManager()
|
||||
|
||||
var received []bus.OutboundMessage
|
||||
ch := &mockChannel{
|
||||
sendFn: func(_ context.Context, msg bus.OutboundMessage) error {
|
||||
received = append(received, msg)
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
w := &channelWorker{
|
||||
ch: ch,
|
||||
limiter: rate.NewLimiter(rate.Inf, 1),
|
||||
}
|
||||
m.channels["test"] = ch
|
||||
m.workers["test"] = w
|
||||
|
||||
msg := bus.OutboundMessage{
|
||||
Channel: "test",
|
||||
ChatID: "123",
|
||||
Content: "hello world",
|
||||
ReplyToMessageID: "msg-456",
|
||||
}
|
||||
|
||||
err := m.SendMessage(context.Background(), msg)
|
||||
if err != nil {
|
||||
t.Fatalf("expected no error, got %v", err)
|
||||
}
|
||||
|
||||
// SendMessage is synchronous — message should already be delivered
|
||||
if len(received) != 1 {
|
||||
t.Fatalf("expected 1 message sent, got %d", len(received))
|
||||
}
|
||||
if received[0].ReplyToMessageID != "msg-456" {
|
||||
t.Fatalf("expected ReplyToMessageID msg-456, got %s", received[0].ReplyToMessageID)
|
||||
}
|
||||
if received[0].Content != "hello world" {
|
||||
t.Fatalf("expected content 'hello world', got %s", received[0].Content)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSendMessage_UnknownChannel(t *testing.T) {
|
||||
m := newTestManager()
|
||||
|
||||
msg := bus.OutboundMessage{
|
||||
Channel: "nonexistent",
|
||||
ChatID: "123",
|
||||
Content: "hello",
|
||||
}
|
||||
|
||||
err := m.SendMessage(context.Background(), msg)
|
||||
if err == nil {
|
||||
t.Fatal("expected error for unknown channel")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSendMessage_NoWorker(t *testing.T) {
|
||||
m := newTestManager()
|
||||
|
||||
ch := &mockChannel{
|
||||
sendFn: func(_ context.Context, _ bus.OutboundMessage) error { return nil },
|
||||
}
|
||||
m.channels["test"] = ch
|
||||
// No worker registered
|
||||
|
||||
msg := bus.OutboundMessage{
|
||||
Channel: "test",
|
||||
ChatID: "123",
|
||||
Content: "hello",
|
||||
}
|
||||
|
||||
err := m.SendMessage(context.Background(), msg)
|
||||
if err == nil {
|
||||
t.Fatal("expected error when no worker exists")
|
||||
}
|
||||
}
|
||||
|
||||
func TestSendMessage_WithRetry(t *testing.T) {
|
||||
m := newTestManager()
|
||||
|
||||
var callCount int
|
||||
ch := &mockChannel{
|
||||
sendFn: func(_ context.Context, _ bus.OutboundMessage) error {
|
||||
callCount++
|
||||
if callCount == 1 {
|
||||
return fmt.Errorf("transient: %w", ErrTemporary)
|
||||
}
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
w := &channelWorker{
|
||||
ch: ch,
|
||||
limiter: rate.NewLimiter(rate.Inf, 1),
|
||||
}
|
||||
m.channels["test"] = ch
|
||||
m.workers["test"] = w
|
||||
|
||||
msg := bus.OutboundMessage{
|
||||
Channel: "test",
|
||||
ChatID: "123",
|
||||
Content: "retry me",
|
||||
}
|
||||
|
||||
err := m.SendMessage(context.Background(), msg)
|
||||
if err != nil {
|
||||
t.Fatalf("expected no error, got %v", err)
|
||||
}
|
||||
|
||||
if callCount != 2 {
|
||||
t.Fatalf("expected 2 Send calls (1 failure + 1 success), got %d", callCount)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSendMessage_WithSplitting(t *testing.T) {
|
||||
m := newTestManager()
|
||||
|
||||
var received []string
|
||||
ch := &mockChannelWithLength{
|
||||
mockChannel: mockChannel{
|
||||
sendFn: func(_ context.Context, msg bus.OutboundMessage) error {
|
||||
received = append(received, msg.Content)
|
||||
return nil
|
||||
},
|
||||
},
|
||||
maxLen: 5,
|
||||
}
|
||||
|
||||
w := &channelWorker{
|
||||
ch: ch,
|
||||
limiter: rate.NewLimiter(rate.Inf, 1),
|
||||
}
|
||||
m.channels["test"] = ch
|
||||
m.workers["test"] = w
|
||||
|
||||
msg := bus.OutboundMessage{
|
||||
Channel: "test",
|
||||
ChatID: "123",
|
||||
Content: "hello world",
|
||||
}
|
||||
|
||||
err := m.SendMessage(context.Background(), msg)
|
||||
if err != nil {
|
||||
t.Fatalf("expected no error, got %v", err)
|
||||
}
|
||||
|
||||
if len(received) < 2 {
|
||||
t.Fatalf("expected message to be split into at least 2 chunks, got %d", len(received))
|
||||
}
|
||||
}
|
||||
|
||||
func TestSendMessage_PreservesOrdering(t *testing.T) {
|
||||
m := newTestManager()
|
||||
|
||||
var order []string
|
||||
ch := &mockChannel{
|
||||
sendFn: func(_ context.Context, msg bus.OutboundMessage) error {
|
||||
order = append(order, msg.Content)
|
||||
return nil
|
||||
},
|
||||
}
|
||||
|
||||
w := &channelWorker{
|
||||
ch: ch,
|
||||
limiter: rate.NewLimiter(rate.Inf, 1),
|
||||
}
|
||||
m.channels["test"] = ch
|
||||
m.workers["test"] = w
|
||||
|
||||
// Send two messages sequentially — they must arrive in order
|
||||
_ = m.SendMessage(context.Background(), bus.OutboundMessage{
|
||||
Channel: "test", ChatID: "1", Content: "first",
|
||||
})
|
||||
_ = m.SendMessage(context.Background(), bus.OutboundMessage{
|
||||
Channel: "test", ChatID: "1", Content: "second",
|
||||
})
|
||||
|
||||
if len(order) != 2 {
|
||||
t.Fatalf("expected 2 messages, got %d", len(order))
|
||||
}
|
||||
if order[0] != "first" || order[1] != "second" {
|
||||
t.Fatalf("expected [first, second], got %v", order)
|
||||
}
|
||||
}
|
||||
|
||||
func TestManager_SendPlaceholder(t *testing.T) {
|
||||
mgr := &Manager{
|
||||
channels: make(map[string]Channel),
|
||||
workers: make(map[string]*channelWorker),
|
||||
placeholders: sync.Map{},
|
||||
}
|
||||
|
||||
mockCh := &mockChannel{
|
||||
sendFn: func(ctx context.Context, msg bus.OutboundMessage) error {
|
||||
return nil
|
||||
},
|
||||
}
|
||||
mgr.channels["mock"] = mockCh
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
// SendPlaceholder should send a placeholder and record it
|
||||
ok := mgr.SendPlaceholder(ctx, "mock", "chat-1")
|
||||
if !ok {
|
||||
t.Fatal("expected SendPlaceholder to succeed")
|
||||
}
|
||||
if mockCh.placeholdersSent != 1 {
|
||||
t.Errorf("expected 1 placeholder sent, got %d", mockCh.placeholdersSent)
|
||||
}
|
||||
|
||||
key := "mock:chat-1"
|
||||
if _, loaded := mgr.placeholders.Load(key); !loaded {
|
||||
t.Error("expected placeholder to be recorded in manager")
|
||||
}
|
||||
|
||||
// SendPlaceholder on unknown channel should return false
|
||||
ok = mgr.SendPlaceholder(ctx, "unknown", "chat-1")
|
||||
if ok {
|
||||
t.Error("expected SendPlaceholder to fail for unknown channel")
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -122,7 +122,11 @@ func (c *SlackChannel) Send(ctx context.Context, msg bus.OutboundMessage) error
|
|||
slack.MsgOptionText(msg.Content, false),
|
||||
}
|
||||
|
||||
if threadTS != "" {
|
||||
if msg.ReplyToMessageID != "" && threadTS == "" {
|
||||
// Answer to the message by creating a Thread under it
|
||||
opts = append(opts, slack.MsgOptionTS(msg.ReplyToMessageID))
|
||||
} else if threadTS != "" {
|
||||
// If we are already in a thread, continue in the thread
|
||||
opts = append(opts, slack.MsgOptionTS(threadTS))
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -182,6 +182,7 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) err
|
|||
// The Manager already splits messages to ≤4000 chars (WithMaxMessageLength),
|
||||
// so msg.Content is guaranteed to be within that limit. We still need to
|
||||
// check if HTML expansion pushes it beyond Telegram's 4096-char API limit.
|
||||
replyToID := msg.ReplyToMessageID
|
||||
queue := []string{msg.Content}
|
||||
for len(queue) > 0 {
|
||||
chunk := queue[0]
|
||||
|
|
@ -202,9 +203,11 @@ func (c *TelegramChannel) Send(ctx context.Context, msg bus.OutboundMessage) err
|
|||
continue
|
||||
}
|
||||
|
||||
if err := c.sendChunk(ctx, chatID, threadID, content, chunk, useMarkdownV2); err != nil {
|
||||
if err := c.sendChunk(ctx, chatID, threadID, content, chunk, replyToID, useMarkdownV2); err != nil {
|
||||
return err
|
||||
}
|
||||
// Only the first chunk should be a reply; subsequent chunks are normal messages.
|
||||
replyToID = ""
|
||||
}
|
||||
|
||||
return nil
|
||||
|
|
@ -216,7 +219,7 @@ func (c *TelegramChannel) sendChunk(
|
|||
ctx context.Context,
|
||||
chatID int64,
|
||||
threadID int,
|
||||
content, mdFallback string,
|
||||
content, replyToID, mdFallback string,
|
||||
useMarkdownV2 bool,
|
||||
) error {
|
||||
tgMsg := tu.Message(tu.ID(chatID), content)
|
||||
|
|
@ -227,6 +230,14 @@ func (c *TelegramChannel) sendChunk(
|
|||
tgMsg.WithParseMode(telego.ModeHTML)
|
||||
}
|
||||
|
||||
if replyToID != "" {
|
||||
if mid, parseErr := strconv.Atoi(replyToID); parseErr == nil {
|
||||
tgMsg.ReplyParameters = &telego.ReplyParameters{
|
||||
MessageID: mid,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if _, err := c.bot.SendMessage(ctx, tgMsg); err != nil {
|
||||
logParseFailed(err, useMarkdownV2)
|
||||
|
||||
|
|
|
|||
|
|
@ -59,6 +59,7 @@ type Config struct {
|
|||
Tools ToolsConfig `json:"tools"`
|
||||
Heartbeat HeartbeatConfig `json:"heartbeat"`
|
||||
Devices DevicesConfig `json:"devices"`
|
||||
Voice VoiceConfig `json:"voice"`
|
||||
// BuildInfo contains build-time version information
|
||||
BuildInfo BuildInfo `json:"build_info,omitempty"`
|
||||
}
|
||||
|
|
@ -473,6 +474,10 @@ type DevicesConfig struct {
|
|||
MonitorUSB bool `json:"monitor_usb" env:"PICOCLAW_DEVICES_MONITOR_USB"`
|
||||
}
|
||||
|
||||
type VoiceConfig struct {
|
||||
EchoTranscription bool `json:"echo_transcription" env:"PICOCLAW_VOICE_ECHO_TRANSCRIPTION"`
|
||||
}
|
||||
|
||||
type ProvidersConfig struct {
|
||||
Anthropic ProviderConfig `json:"anthropic"`
|
||||
OpenAI OpenAIProviderConfig `json:"openai"`
|
||||
|
|
|
|||
|
|
@ -511,6 +511,9 @@ func DefaultConfig() *Config {
|
|||
Enabled: false,
|
||||
MonitorUSB: true,
|
||||
},
|
||||
Voice: VoiceConfig{
|
||||
EchoTranscription: false,
|
||||
},
|
||||
BuildInfo: BuildInfo{
|
||||
Version: Version,
|
||||
GitCommit: GitCommit,
|
||||
|
|
|
|||
|
|
@ -86,14 +86,14 @@ func (s *JSONLStore) metaPath(key string) string {
|
|||
|
||||
// sanitizeKey converts a session key to a safe filename component.
|
||||
// Mirrors pkg/session.sanitizeFilename so that migration paths match.
|
||||
//
|
||||
// Note: this is a lossy mapping — "telegram:123" and "telegram_123"
|
||||
// both produce the same filename. This is an intentional tradeoff:
|
||||
// keys with colons (e.g. from channels) are by far the common case,
|
||||
// and a bidirectional encoding (like URL-encoding) would complicate
|
||||
// file listings and debugging.
|
||||
// Replaces ':' with '_' (session key separator) and '/' and '\' with '_'
|
||||
// so composite IDs (e.g. Telegram forum "chatID/threadID", Slack "channel/thread_ts")
|
||||
// do not create subdirectories or break on Windows.
|
||||
func sanitizeKey(key string) string {
|
||||
return strings.ReplaceAll(key, ":", "_")
|
||||
s := strings.ReplaceAll(key, ":", "_")
|
||||
s = strings.ReplaceAll(s, "/", "_")
|
||||
s = strings.ReplaceAll(s, "\\", "_")
|
||||
return s
|
||||
}
|
||||
|
||||
// readMeta loads the metadata file for a session.
|
||||
|
|
|
|||
|
|
@ -48,6 +48,12 @@ func MigrateFromJSON(
|
|||
if !strings.HasSuffix(name, ".json") {
|
||||
continue
|
||||
}
|
||||
// Skip JSONL metadata files. They are part of the new storage format,
|
||||
// not legacy session snapshots, and re-importing them would overwrite
|
||||
// the paired .jsonl history with an empty message list.
|
||||
if strings.HasSuffix(name, ".meta.json") {
|
||||
continue
|
||||
}
|
||||
// Skip already-migrated files.
|
||||
if strings.HasSuffix(name, ".migrated") {
|
||||
continue
|
||||
|
|
|
|||
|
|
@ -382,3 +382,55 @@ func TestMigrateFromJSON_NonexistentDir(t *testing.T) {
|
|||
t.Errorf("expected 0, got %d", count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMigrateFromJSON_SkipsMetaJSONFiles(t *testing.T) {
|
||||
sessionsDir := t.TempDir()
|
||||
store, err := NewJSONLStore(sessionsDir)
|
||||
if err != nil {
|
||||
t.Fatalf("NewJSONLStore: %v", err)
|
||||
}
|
||||
ctx := context.Background()
|
||||
|
||||
if addErr := store.AddMessage(ctx, "agent:main:pico:direct:pico:test", "user", "keep me"); addErr != nil {
|
||||
t.Fatalf("AddMessage: %v", addErr)
|
||||
}
|
||||
if summaryErr := store.SetSummary(ctx, "agent:main:pico:direct:pico:test", "keep summary"); summaryErr != nil {
|
||||
t.Fatalf("SetSummary: %v", summaryErr)
|
||||
}
|
||||
|
||||
metaPath := filepath.Join(sessionsDir, "agent_main_pico_direct_pico_test.meta.json")
|
||||
if _, statErr := os.Stat(metaPath); statErr != nil {
|
||||
t.Fatalf("meta file missing before migration: %v", statErr)
|
||||
}
|
||||
|
||||
count, err := MigrateFromJSON(ctx, sessionsDir, store)
|
||||
if err != nil {
|
||||
t.Fatalf("MigrateFromJSON: %v", err)
|
||||
}
|
||||
if count != 0 {
|
||||
t.Fatalf("expected 0 migrated, got %d", count)
|
||||
}
|
||||
|
||||
history, err := store.GetHistory(ctx, "agent:main:pico:direct:pico:test")
|
||||
if err != nil {
|
||||
t.Fatalf("GetHistory: %v", err)
|
||||
}
|
||||
if len(history) != 1 || history[0].Content != "keep me" {
|
||||
t.Fatalf("history = %+v, want preserved single message", history)
|
||||
}
|
||||
|
||||
summary, err := store.GetSummary(ctx, "agent:main:pico:direct:pico:test")
|
||||
if err != nil {
|
||||
t.Fatalf("GetSummary: %v", err)
|
||||
}
|
||||
if summary != "keep summary" {
|
||||
t.Fatalf("summary = %q, want %q", summary, "keep summary")
|
||||
}
|
||||
|
||||
if _, statErr := os.Stat(metaPath); statErr != nil {
|
||||
t.Fatalf("meta file should remain in place: %v", statErr)
|
||||
}
|
||||
if _, statErr := os.Stat(metaPath + ".migrated"); !os.IsNotExist(statErr) {
|
||||
t.Fatalf("meta file should not be renamed, stat err = %v", statErr)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -40,6 +40,10 @@ func resolveProviderSelection(cfg *config.Config) (providerSelection, error) {
|
|||
providerName := strings.ToLower(cfg.Agents.Defaults.Provider)
|
||||
lowerModel := strings.ToLower(model)
|
||||
|
||||
if providerName == "" && model == "" {
|
||||
return providerSelection{}, fmt.Errorf("no model configured: agents.defaults.model is empty")
|
||||
}
|
||||
|
||||
sel := providerSelection{
|
||||
providerType: providerTypeHTTPCompat,
|
||||
model: model,
|
||||
|
|
|
|||
|
|
@ -146,12 +146,15 @@ func (sm *SessionManager) TruncateHistory(key string, keepLast int) {
|
|||
}
|
||||
|
||||
// sanitizeFilename converts a session key into a cross-platform safe filename.
|
||||
// Session keys use "channel:chatID" (e.g. "telegram:123456") but ':' is the
|
||||
// volume separator on Windows, so filepath.Base would misinterpret the key.
|
||||
// We replace it with '_'. The original key is preserved inside the JSON file,
|
||||
// so loadSessions still maps back to the right in-memory key.
|
||||
// Replaces ':' with '_' (session key separator) and '/' and '\' with '_' so
|
||||
// composite IDs (e.g. Telegram forum "chatID/threadID") do not create
|
||||
// subdirectories or break on Windows. The original key is preserved inside
|
||||
// the JSON file, so loadSessions still maps back to the right in-memory key.
|
||||
func sanitizeFilename(key string) string {
|
||||
return strings.ReplaceAll(key, ":", "_")
|
||||
s := strings.ReplaceAll(key, ":", "_")
|
||||
s = strings.ReplaceAll(s, "/", "_")
|
||||
s = strings.ReplaceAll(s, "\\", "_")
|
||||
return s
|
||||
}
|
||||
|
||||
func (sm *SessionManager) Save(key string) error {
|
||||
|
|
@ -162,10 +165,9 @@ func (sm *SessionManager) Save(key string) error {
|
|||
filename := sanitizeFilename(key)
|
||||
|
||||
// filepath.IsLocal rejects empty names, "..", absolute paths, and
|
||||
// OS-reserved device names (NUL, COM1 … on Windows).
|
||||
// The extra checks reject "." and any directory separators so that
|
||||
// the session file is always written directly inside sm.storage.
|
||||
if filename == "." || !filepath.IsLocal(filename) || strings.ContainsAny(filename, `/\`) {
|
||||
// OS-reserved device names (NUL, COM1 … on Windows). sanitizeFilename
|
||||
// already replaced '/' and '\' with '_', so no subdirs are created.
|
||||
if filename == "." || !filepath.IsLocal(filename) {
|
||||
return os.ErrInvalid
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@ func TestSanitizeFilename(t *testing.T) {
|
|||
{"slack:C01234", "slack_C01234"},
|
||||
{"no-colons-here", "no-colons-here"},
|
||||
{"multiple:colons:here", "multiple_colons_here"},
|
||||
{"agent:main:telegram:group:-1003822706455/12", "agent_main_telegram_group_-1003822706455_12"},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
|
|
@ -64,11 +65,21 @@ func TestSave_RejectsPathTraversal(t *testing.T) {
|
|||
tmpDir := t.TempDir()
|
||||
sm := NewSessionManager(tmpDir)
|
||||
|
||||
badKeys := []string{"", ".", "..", "foo/bar", "foo\\bar"}
|
||||
// Invalid names that must still be rejected.
|
||||
badKeys := []string{"", ".", ".."}
|
||||
for _, key := range badKeys {
|
||||
sm.GetOrCreate(key)
|
||||
if err := sm.Save(key); err == nil {
|
||||
t.Errorf("Save(%q) should have failed but didn't", key)
|
||||
}
|
||||
}
|
||||
|
||||
// Keys containing path separators are sanitized (no subdirs created).
|
||||
sm.GetOrCreate("foo/bar")
|
||||
if err := sm.Save("foo/bar"); err != nil {
|
||||
t.Fatalf("Save(\"foo/bar\") after sanitize should succeed: %v", err)
|
||||
}
|
||||
if _, err := os.Stat(filepath.Join(tmpDir, "foo_bar.json")); os.IsNotExist(err) {
|
||||
t.Errorf("expected foo_bar.json in storage (sanitized from foo/bar)")
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2,9 +2,18 @@ package utils
|
|||
|
||||
import (
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
"unicode"
|
||||
)
|
||||
|
||||
// Global variable to disable truncation
|
||||
var disableTruncation atomic.Bool
|
||||
|
||||
// SetDisableTruncation globally enables or disables string truncation
|
||||
func SetDisableTruncation(enabled bool) {
|
||||
disableTruncation.Store(enabled)
|
||||
}
|
||||
|
||||
// SanitizeMessageContent removes Unicode control characters, format characters (RTL overrides,
|
||||
// zero-width characters), and other non-graphic characters that could confuse an LLM
|
||||
// or cause display issues in the agent UI.
|
||||
|
|
@ -30,6 +39,10 @@ func SanitizeMessageContent(input string) string {
|
|||
// Handles multi-byte Unicode characters properly.
|
||||
// If the string is truncated, "..." is appended to indicate truncation.
|
||||
func Truncate(s string, maxLen int) string {
|
||||
// If the no-truncate flag is active, it returns the full string
|
||||
if disableTruncation.Load() {
|
||||
return s
|
||||
}
|
||||
if maxLen <= 0 {
|
||||
return ""
|
||||
}
|
||||
|
|
|
|||
|
|
@ -134,6 +134,12 @@ func (h *Handler) startGatewayLocked() (int, error) {
|
|||
execPath := findPicoclawBinary()
|
||||
|
||||
cmd := exec.Command(execPath, "gateway")
|
||||
// Forward the launcher's config path via the environment variable that
|
||||
// GetConfigPath() already reads, so the gateway sub-process uses the same
|
||||
// config file without requiring a --config flag on the gateway subcommand.
|
||||
if h.configPath != "" {
|
||||
cmd.Env = append(os.Environ(), "PICOCLAW_CONFIG="+h.configPath)
|
||||
}
|
||||
|
||||
stdoutPipe, err := cmd.StdoutPipe()
|
||||
if err != nil {
|
||||
|
|
@ -530,18 +536,32 @@ func (h *Handler) currentGatewayStatus() string {
|
|||
}
|
||||
|
||||
// findPicoclawBinary locates the picoclaw executable.
|
||||
// Tries the same directory as the current executable first, then falls back to $PATH.
|
||||
// Search order:
|
||||
// 1. PICOCLAW_BINARY environment variable (explicit override)
|
||||
// 2. Same directory as the current executable
|
||||
// 3. Falls back to "picoclaw" and relies on $PATH
|
||||
func findPicoclawBinary() string {
|
||||
if exe, err := os.Executable(); err == nil {
|
||||
dir := filepath.Dir(exe)
|
||||
candidate := filepath.Join(dir, "picoclaw")
|
||||
binaryName := "picoclaw"
|
||||
if runtime.GOOS == "windows" {
|
||||
candidate += ".exe"
|
||||
binaryName = "picoclaw.exe"
|
||||
}
|
||||
|
||||
// 1. Explicit override via environment variable
|
||||
if p := os.Getenv("PICOCLAW_BINARY"); p != "" {
|
||||
if info, _ := os.Stat(p); info != nil && !info.IsDir() {
|
||||
return p
|
||||
}
|
||||
}
|
||||
|
||||
// 2. Same directory as the launcher executable
|
||||
if exe, err := os.Executable(); err == nil {
|
||||
candidate := filepath.Join(filepath.Dir(exe), binaryName)
|
||||
if info, err := os.Stat(candidate); err == nil && !info.IsDir() {
|
||||
return candidate
|
||||
}
|
||||
}
|
||||
|
||||
// 3. Fall back to PATH lookup
|
||||
return "picoclaw"
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import (
|
|||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
|
@ -120,3 +121,30 @@ func TestGatewayStatusIncludesStartConditionWhenNotReady(t *testing.T) {
|
|||
t.Fatalf("gateway_start_reason missing or not string: %#v", body["gateway_start_reason"])
|
||||
}
|
||||
}
|
||||
|
||||
func TestFindPicoclawBinary_EnvOverride(t *testing.T) {
|
||||
// Create a temporary file to act as the mock binary
|
||||
tmpDir := t.TempDir()
|
||||
mockBinary := filepath.Join(tmpDir, "picoclaw-mock")
|
||||
if err := os.WriteFile(mockBinary, []byte("mock"), 0o755); err != nil {
|
||||
t.Fatalf("WriteFile() error = %v", err)
|
||||
}
|
||||
|
||||
t.Setenv("PICOCLAW_BINARY", mockBinary)
|
||||
|
||||
got := findPicoclawBinary()
|
||||
if got != mockBinary {
|
||||
t.Errorf("findPicoclawBinary() = %q, want %q", got, mockBinary)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFindPicoclawBinary_EnvOverride_InvalidPath(t *testing.T) {
|
||||
// When PICOCLAW_BINARY points to a non-existent path, fall through to next strategy
|
||||
t.Setenv("PICOCLAW_BINARY", "/nonexistent/picoclaw-binary")
|
||||
|
||||
got := findPicoclawBinary()
|
||||
// Should not return the invalid path; falls back to "picoclaw" or another found path
|
||||
if got == "/nonexistent/picoclaw-binary" {
|
||||
t.Errorf("findPicoclawBinary() returned invalid env path %q, expected fallback", got)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue