diff --git a/README.md b/README.md index 5ee9830..59f82b6 100644 --- a/README.md +++ b/README.md @@ -358,6 +358,59 @@ clawlet gateway +
+Feishu (Lark) + +Uses **WebSocket** (no public URL required). clawlet connects to Feishu's WebSocket gateway to receive events in real-time. + +1. Create a Feishu/Lark app + - Go to https://open.feishu.cn/ and create a new enterprise app + - Copy `App ID` (cli_xxx) and `App Secret` from credentials + +2. Configure event subscription + - Enable **Event Subscription** and subscribe to `im.message.receive_v1` + - No webhook URL needed (WebSocket mode) + +3. Add bot permissions + - Bot read/write message permissions + - Install the app to your workspace + +4. Get user OpenID (for allowFrom) + - In Feishu, user IDs are OpenIDs (format: `ou_xxxxx`) + - Empty `allowFrom` allows all users + +5. Configure clawlet + - Set `channels.feishu.enabled=true` + - Configure `appId` and `appSecret` + +Example config (merge into `~/.clawlet/config.json`): + +```json +{ + "channels": { + "feishu": { + "enabled": true, + "appId": "cli_xxxxxxxxxxxxx", + "appSecret": "your-app-secret-here", + "allowFrom": ["ou_xxxxxxxxxxxxx"] + } + } +} +``` + +Then run: + +```bash +clawlet gateway +``` + +Notes: +- The connection is automatic once the app is properly configured +- Both private messages and group messages are supported +- For group messages, the bot needs to be @mentioned by default + +
+ ## CLI Reference | Command | Description | diff --git a/agent/agent.go b/agent/agent.go index e476460..39f9d0e 100644 --- a/agent/agent.go +++ b/agent/agent.go @@ -80,6 +80,7 @@ func New(opts Options) (*Agent, error) { MaxTokens: opts.Config.Agents.Defaults.MaxTokensValue(), Temperature: opts.Config.Agents.Defaults.Temperature, Headers: opts.Config.LLM.Headers, + LogLevel: opts.Config.Log.LevelValue(), } treg := &tools.Registry{ diff --git a/agent/loop.go b/agent/loop.go index b673006..6ae383f 100644 --- a/agent/loop.go +++ b/agent/loop.go @@ -94,6 +94,7 @@ func NewLoop(opts LoopOptions) (*Loop, error) { MaxTokens: opts.Config.Agents.Defaults.MaxTokensValue(), Temperature: opts.Config.Agents.Defaults.Temperature, Headers: opts.Config.LLM.Headers, + LogLevel: opts.Config.Log.LevelValue(), } treg := &tools.Registry{ diff --git a/bus/bus.go b/bus/bus.go index c7c983a..c32ccb3 100644 --- a/bus/bus.go +++ b/bus/bus.go @@ -49,11 +49,12 @@ type InboundMessage struct { } type OutboundMessage struct { - Channel string - ChatID string - Content string - ReplyTo string - Delivery Delivery + Channel string + ChatID string + Content string + ReplyTo string + Delivery Delivery + Attachments []Attachment } type Bus struct { diff --git a/channels/discord/discord.go b/channels/discord/discord.go index 061882b..b479d75 100644 --- a/channels/discord/discord.go +++ b/channels/discord/discord.go @@ -1,12 +1,14 @@ package discord import ( + "bytes" "context" "errors" "fmt" "log" "net" "net/http" + "os" "strings" "sync" "sync/atomic" @@ -103,7 +105,8 @@ func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { return fmt.Errorf("chat_id is empty") } content := strings.TrimSpace(msg.Content) - if content == "" { + attachments := msg.Attachments + if content == "" && len(attachments) == 0 { return nil } @@ -125,7 +128,7 @@ func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { replyToID := resolveDiscordReplyTarget(msg) const maxAttempts = 3 for attempt := 1; attempt <= maxAttempts; attempt++ { - err := sendDiscordMessage(dg, chID, content, replyToID) + err := sendDiscordMessage(dg, chID, content, replyToID, attachments) if err == nil { return nil } @@ -233,21 +236,43 @@ func buildDiscordDelivery(m *discordgo.MessageCreate) bus.Delivery { return d } -func sendDiscordMessage(dg *discordgo.Session, chID, content, replyToID string) error { - if replyToID == "" { - _, err := dg.ChannelMessageSend(chID, content) - return err - } - _, err := dg.ChannelMessageSendComplex(chID, &discordgo.MessageSend{ +func sendDiscordMessage(dg *discordgo.Session, chID, content, replyToID string, attachments []bus.Attachment) error { + msgSend := &discordgo.MessageSend{ Content: content, - Reference: &discordgo.MessageReference{ + } + + // Add reply reference if needed + if replyToID != "" { + msgSend.Reference = &discordgo.MessageReference{ MessageID: replyToID, ChannelID: chID, - }, - AllowedMentions: &discordgo.MessageAllowedMentions{ + } + msgSend.AllowedMentions = &discordgo.MessageAllowedMentions{ RepliedUser: false, - }, + } + } + + // Add attachments + for _, att := range attachments { + if att.Data != nil { + msgSend.Files = append(msgSend.Files, &discordgo.File{ + Name: att.Name, + Reader: bytes.NewReader(att.Data), }) + } else if att.LocalPath != "" { + file, err := os.Open(att.LocalPath) + if err != nil { + return err +} + defer file.Close() + msgSend.Files = append(msgSend.Files, &discordgo.File{ + Name: att.Name, + Reader: file, + }) + } + } + + _, err := dg.ChannelMessageSendComplex(chID, msgSend) return err } diff --git a/channels/feishu/feishu.go b/channels/feishu/feishu.go new file mode 100644 index 0000000..e6c1679 --- /dev/null +++ b/channels/feishu/feishu.go @@ -0,0 +1,1445 @@ +package feishu + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "io" + "log" + "mime/multipart" + "net/http" + "net/url" + "os" + "path/filepath" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/gorilla/websocket" + "github.com/larksuite/oapi-sdk-go/v3/ws" + "github.com/mosaxiv/clawlet/bus" + "github.com/mosaxiv/clawlet/channels" + "github.com/mosaxiv/clawlet/config" +) + +// Feishu API constants +const ( + feishuAPIBase = "https://open.feishu.cn" +) + +// Event payload structure for im.message.receive_v1/v2 +// V2 format: {"header":{...}, "event":{"sender":{...}, "message":{...}}} +// V1 format: {"header":{...}, "event":{"sender_id":{...}, "sender_type":"...", "message":{...}}} +type feishuEventPayload struct { + Schema string `json:"schema"` + Header *feishuEventHeader `json:"header"` + Event *feishuEventDataV1 `json:"event"` +} + +type feishuEventHeader struct { + EventType string `json:"event_type"` + EventID string `json:"event_id"` + AppID string `json:"app_id"` + TenantKey string `json:"tenant_key"` + Timestamp string `json:"timestamp"` +} + +// V1 event data format (direct fields) +type feishuEventDataV1 struct { + SenderID *feishuUserID `json:"sender_id,omitempty"` + SenderType *string `json:"sender_type,omitempty"` + TenantKey *string `json:"tenant_key,omitempty"` + Message *feishuEventMessage `json:"message,omitempty"` + // V2 format fields (nested) + Sender *feishuEventSender `json:"sender,omitempty"` +} + +type feishuEventSender struct { + SenderID *feishuUserID `json:"sender_id,omitempty"` + SenderType *string `json:"sender_type,omitempty"` + TenantKey *string `json:"tenant_key,omitempty"` +} + +type feishuUserID struct { + UserId *string `json:"user_id,omitempty"` + OpenId *string `json:"open_id,omitempty"` + UnionId *string `json:"union_id,omitempty"` +} + +type feishuEventMessage struct { + MessageID *string `json:"message_id,omitempty"` + RootID *string `json:"root_id,omitempty"` + ParentID *string `json:"parent_id,omitempty"` + ChatID *string `json:"chat_id,omitempty"` + MsgType *string `json:"message_type,omitempty"` // v1 uses "message_type" + Content *string `json:"content,omitempty"` + SenderID *feishuUserID `json:"sender_id,omitempty"` + SenderType *string `json:"sender_type,omitempty"` + CreateTime *string `json:"create_time,omitempty"` + UpdateID *int64 `json:"update_id,omitempty"` + ChatType *string `json:"chat_type,omitempty"` + MentionAll *bool `json:"mention_all,omitempty"` +} + +// Legacy structure for direct message parsing +type feishuMessageV2 struct { + MessageID string `json:"message_id"` + RootID string `json:"root_id"` + ParentID string `json:"parent_id"` + ChatID string `json:"chat_id"` + MsgType string `json:"msg_type"` + Content string `json:"content"` + SenderID string `json:"sender_id"` + SenderType string `json:"sender_type"` + SenderTenantKey string `json:"sender_tenant_key"` + CreateTime string `json:"create_time"` + UpdateID int64 `json:"update_id"` + ChatType string `json:"chat_type"` + MentionAll *bool `json:"mention_all"` +} + +type feishuMessageContent struct { + Text string `json:"text"` +} + +type Channel struct { + cfg config.FeishuConfig + bus *bus.Bus + allow channels.AllowList + + running atomic.Bool + + mu sync.Mutex + conn *websocket.Conn + client *http.Client + appID string + appSecret string + cancel context.CancelFunc + + // Bot info + botOpenID string + + // Processed message dedup + processedIDs map[string]bool + processedMu sync.Mutex + + // Connection info + connURL *url.URL + connID string + serviceID string + + // Config + reconnectCount int + reconnectInterval time.Duration + pingInterval time.Duration +} + +func New(cfg config.FeishuConfig, b *bus.Bus) *Channel { + return &Channel{ + cfg: cfg, + bus: b, + allow: channels.AllowList{AllowFrom: cfg.AllowFrom}, + client: &http.Client{Timeout: 30 * time.Second}, + processedIDs: make(map[string]bool), + reconnectCount: -1, // unlimited + reconnectInterval: 120 * time.Second, + pingInterval: 120 * time.Second, + } +} + +func (c *Channel) Name() string { return "feishu" } +func (c *Channel) IsRunning() bool { return c.running.Load() } + +func (c *Channel) Start(ctx context.Context) error { + if strings.TrimSpace(c.cfg.AppID) == "" { + return fmt.Errorf("feishu appID is empty") + } + if strings.TrimSpace(c.cfg.AppSecret) == "" { + return fmt.Errorf("feishu appSecret is empty") + } + + runCtx, cancel := context.WithCancel(ctx) + defer cancel() + + c.mu.Lock() + c.appID = strings.TrimSpace(c.cfg.AppID) + c.appSecret = strings.TrimSpace(c.cfg.AppSecret) + c.cancel = cancel + c.mu.Unlock() + + c.running.Store(true) + defer c.running.Store(false) + + // Fetch bot open_id + c.fetchBotOpenID(runCtx) + + // Connect to WebSocket + if err := c.connect(runCtx); err != nil { + return fmt.Errorf("failed to connect to feishu websocket: %w", err) + } + + log.Printf("feishu channel started (app_id=%s)\n", c.appID[:min(12, len(c.appID))]) + + return runCtx.Err() +} + +func (c *Channel) Stop() error { + c.running.Store(false) + c.mu.Lock() + cancel := c.cancel + c.cancel = nil + c.mu.Unlock() + + if cancel != nil { + cancel() + } + + c.mu.Lock() + conn := c.conn + c.conn = nil + c.mu.Unlock() + + if conn != nil { + conn.Close() + } + return nil +} + +func (c *Channel) connect(ctx context.Context) error { + // Get WebSocket URL + connURL, err := c.getConnURL(ctx) + if err != nil { + return fmt.Errorf("failed to get connection URL: %w", err) + } + + u, err := url.Parse(connURL) + if err != nil { + return fmt.Errorf("failed to parse URL: %w", err) + } + + connID := u.Query().Get("device_id") + serviceID := u.Query().Get("service_id") + + log.Printf("feishu: got websocket endpoint (conn_id=%s, service_id=%s)\n", connID, serviceID) + + // Connect + conn, _, err := websocket.DefaultDialer.Dial(connURL, nil) + if err != nil { + return err + } + + c.mu.Lock() + c.conn = conn + c.connURL = u + c.connID = connID + c.serviceID = serviceID + c.mu.Unlock() + + log.Println("feishu: websocket connected") + + // Start ping loop + go c.pingLoop(ctx) + + // Run receive loop in current goroutine (don't return) + c.receiveMessageLoop(ctx) + + return nil +} + +func (c *Channel) getConnURL(ctx context.Context) (string, error) { + body := map[string]string{ + "AppID": c.appID, + "AppSecret": c.appSecret, + } + bs, _ := json.Marshal(body) + + req, _ := http.NewRequestWithContext(ctx, http.MethodPost, feishuAPIBase+"/callback/ws/endpoint", bytes.NewBuffer(bs)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("locale", "zh") + + resp, err := c.client.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + + var result struct { + Code int `json:"code"` + Msg string `json:"msg"` + Data *struct { + URL string `json:"url"` + ClientConfig *struct { + ReconnectCount int `json:"reconnect_count"` + ReconnectInterval int `json:"reconnect_interval"` + ReconnectNonce int `json:"reconnect_nonce"` + PingInterval int `json:"ping_interval"` + } `json:"client_config"` + } `json:"data"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return "", err + } + + if result.Code != 0 { + return "", fmt.Errorf("feishu endpoint error: code=%d, msg=%s", result.Code, result.Msg) + } + + if result.Data == nil || result.Data.URL == "" { + return "", fmt.Errorf("feishu endpoint returned empty URL") + } + + if result.Data.ClientConfig != nil { + c.reconnectCount = result.Data.ClientConfig.ReconnectCount + c.reconnectInterval = time.Duration(result.Data.ClientConfig.ReconnectInterval) * time.Second + c.pingInterval = time.Duration(result.Data.ClientConfig.PingInterval) * time.Second + } + + return result.Data.URL, nil +} + +func (c *Channel) receiveMessageLoop(ctx context.Context) { + for { + select { + case <-ctx.Done(): + log.Println("feishu: receiveMessageLoop: context done") + return + default: + } + + c.mu.Lock() + conn := c.conn + c.mu.Unlock() + + if conn == nil { + log.Println("feishu: receiveMessageLoop: connection is nil") + return + } + + mt, msg, err := conn.ReadMessage() + if err != nil { + log.Printf("feishu: receiveMessageLoop: read error: %v\n", err) + c.disconnect(ctx) + go c.reconnect(ctx) + return + } + + log.Printf("feishu: <<< received WS message: type=%d, len=%d\n", mt, len(msg)) + + if mt != websocket.BinaryMessage { + log.Printf("feishu: received non-binary message type: %d\n", mt) + continue + } + + go c.handleMessage(ctx, msg) + } +} + +func (c *Channel) handleMessage(ctx context.Context, msg []byte) { + defer func() { + if r := recover(); r != nil { + log.Printf("feishu: handleMessage panic: %v\n", r) + } + }() + + log.Printf("feishu: handling message (%d bytes): % x\n", len(msg), msg[:min(32, len(msg))]) + + // Parse frame using official SDK + var frame ws.Frame + if err := frame.Unmarshal(msg); err != nil { + log.Printf("feishu: unmarshal frame failed: %v\n", err) + return + } + + log.Printf("feishu: parsed frame: method=%d, service=%d, len(headers)=%d, len(payload)=%d\n", + frame.Method, frame.Service, len(frame.Headers), len(frame.Payload)) + + // Handle based on frame type + if frame.Method == 0 { // CONTROL + log.Println("feishu: handling CONTROL frame") + c.handleControlFrame(ctx, frame) + } else if frame.Method == 1 { // DATA + log.Println("feishu: handling DATA frame") + c.handleDataFrame(ctx, frame) + } else { + log.Printf("feishu: unknown frame method: %d\n", frame.Method) + } +} + +func (c *Channel) handleControlFrame(ctx context.Context, frame ws.Frame) { + hs := ws.Headers(frame.Headers) + t := hs.GetString("type") + + if t == "pong" { + log.Println("feishu: received pong") + if len(frame.Payload) > 0 { + var conf struct { + PingInterval int `json:"ping_interval"` + ReconnectCount int `json:"reconnect_count"` + ReconnectInterval int `json:"reconnect_interval"` + } + if err := json.Unmarshal(frame.Payload, &conf); err == nil { + if conf.PingInterval > 0 { + c.pingInterval = time.Duration(conf.PingInterval) * time.Second + } + } + } + } +} + +func (c *Channel) handleDataFrame(ctx context.Context, frame ws.Frame) { + hs := ws.Headers(frame.Headers) + msgID := hs.GetString("message_id") + traceID := hs.GetString("trace_id") + type_ := hs.GetString("type") + + log.Printf("feishu: received data frame: type=%s, msg_id=%s, trace_id=%s\n", type_, msgID, traceID) + + if type_ != "event" { + log.Printf("feishu: skipping non-event frame\n") + return + } + + // Parse event payload + var payload feishuEventPayload + if err := json.Unmarshal(frame.Payload, &payload); err != nil { + log.Printf("feishu: unmarshal event payload failed: %v\n", err) + return + } + + log.Printf("feishu: event type: %s\n", payload.Header.EventType) + + // Handle message receive events + if payload.Header.EventType == "im.message.receive_v1" || payload.Header.EventType == "im.message.receive_v2" { + log.Printf("feishu: processing %s event\n", payload.Header.EventType) + + if payload.Event == nil { + log.Println("feishu: event is nil") + return + } + if payload.Event.Message == nil { + log.Println("feishu: event.Message is nil") + return + } + msg := payload.Event.Message + + // Get sender ID from event level (v1) or message level (v2) + var senderIDStr string + if payload.Event.SenderID != nil && payload.Event.SenderID.OpenId != nil { + senderIDStr = *payload.Event.SenderID.OpenId + } else if msg.SenderID != nil && msg.SenderID.OpenId != nil { + senderIDStr = *msg.SenderID.OpenId + } + + // Safe access with nil checks + chatID := "" + if msg.ChatID != nil { + chatID = *msg.ChatID + } + msgType := "" + if msg.MsgType != nil { + msgType = *msg.MsgType + } + content := "" + if msg.Content != nil { + content = *msg.Content + } + chatType := "" + if msg.ChatType != nil { + chatType = *msg.ChatType + } + + log.Printf("feishu: received message: chatID=%s, senderID=%s, msgType=%s, chatType=%s, content=%s\n", + chatID, senderIDStr, msgType, chatType, content) + + // Process the message + c.processMessageV2(ctx, msg, frame.Payload, hs, senderIDStr) + } else { + log.Printf("feishu: unhandled event type: %s\n", payload.Header.EventType) + } +} + +func (c *Channel) processMessageV2(ctx context.Context, msg *feishuEventMessage, rawPayload []byte, hs ws.Headers, senderID string) { + if msg == nil { + log.Println("feishu: message is nil") + return + } + + // Extract values with nil checks + messageID := "" + if msg.MessageID != nil { + messageID = *msg.MessageID + } + chatID := "" + if msg.ChatID != nil { + chatID = *msg.ChatID + } + msgType := "" + if msg.MsgType != nil { + msgType = *msg.MsgType + } + content := "" + if msg.Content != nil { + content = *msg.Content + } + chatType := "" + if msg.ChatType != nil { + chatType = *msg.ChatType + } + rootID := "" + if msg.RootID != nil { + rootID = *msg.RootID + } + parentID := "" + if msg.ParentID != nil { + parentID = *msg.ParentID + } + + log.Printf("feishu: processing message: ID=%s, chatID=%s, senderID=%s, msgType=%s\n", messageID, chatID, senderID, msgType) + + // Dedup check + c.processedMu.Lock() + if c.processedIDs[messageID] { + c.processedMu.Unlock() + log.Printf("feishu: duplicate message %s, skipping\n", messageID) + return + } + c.processedIDs[messageID] = true + if len(c.processedIDs) > 1000 { + for k := range c.processedIDs { + delete(c.processedIDs, k) + if len(c.processedIDs) <= 500 { + break + } + } + } + c.processedMu.Unlock() + + // Ignore bot's own messages + if c.botOpenID != "" && senderID == c.botOpenID { + log.Println("feishu: ignoring bot's own message") + return + } + + // Handle different message types + isDirect := chatType == "p2p" || strings.HasPrefix(chatID, "oc_") + switch msgType { + case "text": + c.processTextMessage(ctx, content, senderID, chatID, messageID, rootID, parentID, isDirect) + case "image": + c.processImageMessage(ctx, content, senderID, chatID, messageID, rootID, parentID, isDirect) + case "file": + c.processFileMessage(ctx, content, senderID, chatID, messageID, rootID, parentID, isDirect) + default: + log.Printf("feishu: unsupported message type: %s\n", msgType) + } +} + +func (c *Channel) processTextMessage(ctx context.Context, content, senderID, chatID, messageID, rootID, parentID string, isDirect bool) { + var msgContent feishuMessageContent + if err := json.Unmarshal([]byte(content), &msgContent); err != nil { + log.Printf("feishu: failed to parse content: %v\n", err) + return + } + + text := strings.TrimSpace(msgContent.Text) + log.Printf("feishu: message text: %s\n", text) + if text == "" { + log.Println("feishu: empty message text") + return + } + + // Check allowlist + if !c.allow.Allowed(senderID) { + log.Printf("feishu: sender %s not in allowlist\n", senderID) + return + } + + // Strip bot @mention + text = c.stripBotMention(text) + log.Printf("feishu: after strip mention: %s\n", text) + if text == "" { + log.Println("feishu: empty text after stripping mentions") + return + } + + // Publish to bus + log.Println("feishu: publishing to bus") + _ = c.bus.PublishInbound(ctx, bus.InboundMessage{ + Channel: "feishu", + SenderID: senderID, + ChatID: chatID, + Content: text, + SessionKey: "feishu:" + chatID, + Delivery: bus.Delivery{ + MessageID: messageID, + ThreadID: rootID, + ReplyToID: parentID, + IsDirect: isDirect, + }, + }) + + // Send acknowledgment response to Feishu + c.sendEventAck(ctx, messageID, "") +} + +func (c *Channel) processImageMessage(ctx context.Context, content, senderID, chatID, messageID, rootID, parentID string, isDirect bool) { + // Parse image content - format: {"image_key":"img_xxx"} + var imgContent struct { + ImageKey string `json:"image_key"` + } + if err := json.Unmarshal([]byte(content), &imgContent); err != nil { + log.Printf("feishu: failed to parse image content: %v, raw: %s\n", err, content) + return + } + + if imgContent.ImageKey == "" { + log.Println("feishu: empty image_key") + return + } + + log.Printf("feishu: received image: image_key=%s\n", imgContent.ImageKey) + + // Check allowlist + if !c.allow.Allowed(senderID) { + log.Printf("feishu: sender %s not in allowlist\n", senderID) + return + } + + // Download image to local workspace (optional - continue even if download fails) + localPath := "" + downloadedPath, err := c.downloadImage(ctx, imgContent.ImageKey, chatID) + if err != nil { + log.Printf("feishu: failed to download image: %v (continuing without local file)\n", err) + } else { + localPath = downloadedPath + } + + _ = c.bus.PublishInbound(ctx, bus.InboundMessage{ + Channel: "feishu", + SenderID: senderID, + ChatID: chatID, + Content: "[Image]", + Attachments: []bus.Attachment{ + { + ID: imgContent.ImageKey, + Name: "image.jpg", + MIMEType: "image/jpeg", + Kind: "image", + LocalPath: localPath, + }, + }, + SessionKey: "feishu:" + chatID, + Delivery: bus.Delivery{ + MessageID: messageID, + ThreadID: rootID, + ReplyToID: parentID, + IsDirect: isDirect, + }, + }) + + // Send acknowledgment response to Feishu + c.sendEventAck(ctx, messageID, "") + log.Println("feishu: image message published to bus") +} + +func (c *Channel) processFileMessage(ctx context.Context, content, senderID, chatID, messageID, rootID, parentID string, isDirect bool) { + var fileContent struct { + FileKey string `json:"file_key"` + FileName string `json:"file_name"` + } + if err := json.Unmarshal([]byte(content), &fileContent); err != nil { + log.Printf("feishu: failed to parse file content: %v\n", err) + return + } + + if fileContent.FileKey == "" { + log.Println("feishu: empty file_key") + return + } + + log.Printf("feishu: received file: file_key=%s\n", fileContent.FileKey) + + // Check allowlist + if !c.allow.Allowed(senderID) { + log.Printf("feishu: sender %s not in allowlist\n", senderID) + return + } + + // Download file to local workspace (optional - continue even if download fails) + localPath := "" + downloadedPath, err := c.downloadFile(ctx, fileContent.FileKey, chatID, fileContent.FileName) + if err != nil { + log.Printf("feishu: failed to download file: %v (continuing without local file)\n", err) + } else { + localPath = downloadedPath + } + + fileName := fileContent.FileName + if fileName == "" { + fileName = "file" + } + + _ = c.bus.PublishInbound(ctx, bus.InboundMessage{ + Channel: "feishu", + SenderID: senderID, + ChatID: chatID, + Content: "[File]", + Attachments: []bus.Attachment{ + { + ID: fileContent.FileKey, + Name: fileName, + MIMEType: "application/octet-stream", + Kind: "file", + LocalPath: localPath, + }, + }, + SessionKey: "feishu:" + chatID, + Delivery: bus.Delivery{ + MessageID: messageID, + ThreadID: rootID, + ReplyToID: parentID, + IsDirect: isDirect, + }, + }) + + // Send acknowledgment response to Feishu + c.sendEventAck(ctx, messageID, "") + log.Println("feishu: file message published to bus") +} + +func (c *Channel) publishMessageToBus(ctx context.Context, senderID, chatID, text, messageID, rootID, parentID string, isDirect bool) { + // Check allowlist + if !c.allow.Allowed(senderID) { + log.Printf("feishu: sender %s not in allowlist\n", senderID) + return + } + + // Strip bot @mention + text = c.stripBotMention(text) + log.Printf("feishu: after strip mention: %s\n", text) + if text == "" { + log.Println("feishu: empty text after stripping mentions") + return + } + + // Determine if direct message + isDirect = isDirect || strings.HasPrefix(chatID, "oc_") + log.Printf("feishu: isDirect=%v\n", isDirect) + + // Publish to bus + log.Println("feishu: publishing to bus") + _ = c.bus.PublishInbound(ctx, bus.InboundMessage{ + Channel: "feishu", + SenderID: senderID, + ChatID: chatID, + Content: text, + SessionKey: "feishu:" + chatID, + Delivery: bus.Delivery{ + MessageID: messageID, + ThreadID: rootID, + ReplyToID: parentID, + IsDirect: isDirect, + }, + }) +} + +// sendEventAck sends an acknowledgment response to Feishu for event processing +func (c *Channel) sendEventAck(ctx context.Context, msgID, traceID string) { + c.mu.Lock() + conn := c.conn + serviceID := c.serviceID + c.mu.Unlock() + + if conn == nil || serviceID == "" { + return + } + + sid := int32(0) + _, _ = fmt.Sscanf(serviceID, "%d", &sid) + + // Build response frame + response := map[string]interface{}{ + "code": 0, // HTTP 200 OK + } + responseJSON, _ := json.Marshal(response) + + frame := &ws.Frame{ + SeqID: 0, + LogID: 0, + Service: sid, + Method: 1, // DATA frame + Headers: []ws.Header{ + {Key: "message_id", Value: msgID}, + {Key: "trace_id", Value: traceID}, + {Key: "type", Value: "event"}, + }, + Payload: responseJSON, + } + + data, _ := frame.Marshal() + if err := conn.WriteMessage(websocket.BinaryMessage, data); err != nil { + log.Printf("feishu: failed to send event ack: %v\n", err) + } else { + log.Println("feishu: sent event acknowledgment") + } +} + +func (c *Channel) stripBotMention(text string) string { + text = strings.TrimSpace(text) + + for { + start := strings.Index(text, "") + if end == -1 { + break + } + end += start + 5 + mention := text[start:end] + + if strings.Contains(mention, `user_id="`) { + text = strings.Replace(text, mention, "", 1) + text = strings.TrimSpace(text) + text = strings.TrimLeft(text, ":,，.。:：") + text = strings.TrimSpace(text) + continue + } + break + } + return text +} + +func (c *Channel) pingLoop(ctx context.Context) { + defer func() { + if r := recover(); r != nil { + log.Printf("feishu: ping loop panic: %v\n", r) + } + }() + + for { + select { + case <-ctx.Done(): + return + case <-time.After(c.pingInterval): + c.sendPing(ctx) + } + } +} + +func (c *Channel) sendPing(ctx context.Context) { + c.mu.Lock() + conn := c.conn + serviceID := c.serviceID + c.mu.Unlock() + + if conn == nil || serviceID == "" { + return + } + + i := int32(0) + _, _ = fmt.Sscanf(serviceID, "%d", &i) + + frame := ws.NewPingFrame(i) + bs, _ := frame.Marshal() + + if err := conn.WriteMessage(websocket.BinaryMessage, bs); err != nil { + log.Printf("feishu: ping failed: %v\n", err) + } else { + log.Println("feishu: sent ping") + } +} + +func (c *Channel) reconnect(ctx context.Context) { + if !c.running.Load() { + return + } + + log.Println("feishu: attempting to reconnect...") + + // Random nonce for first reconnect + if c.reconnectCount < 0 || c.reconnectCount > 0 { + time.Sleep(time.Duration(randInt(0, 30000)) * time.Millisecond) + } + + count := 0 + for c.reconnectCount < 0 || count < c.reconnectCount { + if err := c.connect(ctx); err != nil { + log.Printf("feishu: reconnect failed: %v\n", err) + count++ + time.Sleep(c.reconnectInterval) + continue + } + return + } + + log.Printf("feishu: reconnect failed after %d attempts\n", count) +} + +func (c *Channel) disconnect(ctx context.Context) { + c.mu.Lock() + defer c.mu.Unlock() + + if c.conn == nil { + return + } + + c.conn.Close() + log.Printf("feishu: disconnected from %s\n", c.connURL) + + c.conn = nil + c.connURL = nil + c.connID = "" + c.serviceID = "" +} + +func (c *Channel) getImageURL(ctx context.Context, imageKey string) (string, error) { + token, err := c.getTenantAccessToken(ctx) + if err != nil { + return "", err + } + + // Get image download URL + url := fmt.Sprintf("%s/open-apis/im/v1/images/%s", feishuAPIBase, imageKey) + req, _ := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + req.Header.Set("Authorization", "Bearer "+token) + + resp, err := c.client.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + + var result struct { + Code int `json:"code"` + Msg string `json:"msg"` + Data struct { + ImageKey string `json:"image_key"` + ImageURL string `json:"image_url"` + } `json:"data"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return "", err + } + + if result.Code != 0 { + return "", fmt.Errorf("feishu API error: code=%d, msg=%s", result.Code, result.Msg) + } + + return result.Data.ImageURL, nil +} + +func (c *Channel) downloadImage(ctx context.Context, imageKey, chatID string) (string, error) { + token, err := c.getTenantAccessToken(ctx) + if err != nil { + return "", err + } + + // Get image download URL + url := fmt.Sprintf("%s/open-apis/im/v1/images/%s", feishuAPIBase, imageKey) + req, _ := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + req.Header.Set("Authorization", "Bearer "+token) + + resp, err := c.client.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return "", fmt.Errorf("failed to download image: status=%d", resp.StatusCode) + } + + // Read image data + data, err := io.ReadAll(resp.Body) + if err != nil { + return "", err + } + + // Save to workspace directory + filename := fmt.Sprintf("feishu_image_%s_%s.jpg", chatID, imageKey) + localPath, err := c.saveToWorkspace(filename, data) + if err != nil { + return "", err + } + + log.Printf("feishu: image saved to %s\n", localPath) + return localPath, nil +} + +func (c *Channel) downloadFile(ctx context.Context, fileKey, chatID, fileName string) (string, error) { + token, err := c.getTenantAccessToken(ctx) + if err != nil { + return "", err + } + + // Get file download URL + url := fmt.Sprintf("%s/open-apis/im/v1/files/%s", feishuAPIBase, fileKey) + req, _ := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + req.Header.Set("Authorization", "Bearer "+token) + + resp, err := c.client.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + return "", fmt.Errorf("failed to download file: status=%d", resp.StatusCode) + } + + // Read file data + data, err := io.ReadAll(resp.Body) + if err != nil { + return "", err + } + + // Save to workspace directory + if fileName == "" { + fileName = fmt.Sprintf("feishu_file_%s_%s", chatID, fileKey) + } else { + fileName = fmt.Sprintf("feishu_%s_%s", chatID, fileName) + } + localPath, err := c.saveToWorkspace(fileName, data) + if err != nil { + return "", err + } + + log.Printf("feishu: file saved to %s\n", localPath) + return localPath, nil +} + +func (c *Channel) saveToWorkspace(filename string, data []byte) (string, error) { + // Get workspace directory (use current directory or CLAWLET_WORKSPACE env) + workspace := os.Getenv("CLAWLET_WORKSPACE") + if workspace == "" { + workspace = "." + } + + // Create feishu subdirectory + feishuDir := filepath.Join(workspace, "feishu_files") + if err := os.MkdirAll(feishuDir, 0755); err != nil { + return "", err + } + + // Sanitize filename + filename = strings.ReplaceAll(filename, "/", "_") + filename = strings.ReplaceAll(filename, "\\", "_") + filename = strings.ReplaceAll(filename, "..", "_") + + localPath := filepath.Join(feishuDir, filename) + if err := os.WriteFile(localPath, data, 0644); err != nil { + return "", err + } + + return localPath, nil +} + +func (c *Channel) fetchBotOpenID(ctx context.Context) { + token, err := c.getTenantAccessToken(ctx) + if err != nil { + log.Printf("feishu: failed to get token for bot info: %v\n", err) + return + } + + req, _ := http.NewRequestWithContext(ctx, http.MethodGet, feishuAPIBase+"/open-apis/bot/v3/info", nil) + req.Header.Set("Authorization", "Bearer "+token) + + resp, err := c.client.Do(req) + if err != nil { + return + } + defer resp.Body.Close() + + var result struct { + Code int `json:"code"` + Msg string `json:"msg"` + Bot struct { + OpenID string `json:"open_id"` + } `json:"bot"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return + } + + if result.Code == 0 && result.Bot.OpenID != "" { + c.botOpenID = result.Bot.OpenID + log.Printf("feishu: bot open_id=%s\n", c.botOpenID[:min(12, len(c.botOpenID))]) + } +} + +func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { + log.Printf("feishu: Send called: channel=%s, chatID=%s, content=%s, attachments=%d\n", + msg.Channel, msg.ChatID, msg.Content, len(msg.Attachments)) + + text := strings.TrimSpace(msg.Content) + attachments := msg.Attachments + + // If only attachments, send them + if text == "" && len(attachments) > 0 { + log.Printf("feishu: sending only attachments\n") + for i, att := range attachments { + log.Printf("feishu: attachment[%d]: name=%s, kind=%s, localPath=%s, hasData=%v\n", + i, att.Name, att.Kind, att.LocalPath, att.Data != nil) + if err := c.sendAttachment(ctx, msg.ChatID, att, msg); err != nil { + log.Printf("feishu: failed to send attachment: %v\n", err) + return err + } + } + return nil + } + + // If only text, send text + if text != "" && len(attachments) == 0 { + log.Printf("feishu: sending only text\n") + return c.sendTextMessage(ctx, msg.ChatID, text, msg) + } + + // If both text and attachments, send text first then attachments + if text != "" { + log.Printf("feishu: sending text first\n") + if err := c.sendTextMessage(ctx, msg.ChatID, text, msg); err != nil { + return err + } + } + + for i, att := range attachments { + log.Printf("feishu: sending attachment[%d]: name=%s, kind=%s\n", i, att.Name, att.Kind) + if err := c.sendAttachment(ctx, msg.ChatID, att, msg); err != nil { + log.Printf("feishu: failed to send attachment: %v\n", err) + return err + } + } + + return nil +} + +func (c *Channel) sendTextMessage(ctx context.Context, chatID, text string, msg bus.OutboundMessage) error { + token, err := c.getTenantAccessToken(ctx) + if err != nil { + return err + } + + // Build message content - text message format + content := map[string]string{"text": text} + contentJSON, _ := json.Marshal(content) + + // Determine receiver type and ID + receiverType, receiveID := c.resolveReceiverType(chatID) + + // Build request body + body := map[string]interface{}{ + "receive_id": receiveID, + "msg_type": "text", + "content": string(contentJSON), + } + + // Add reply_id if present (use message_id for reply) + replyTo := c.resolveReplyTo(msg) + if replyTo != "" { + body["reply_id"] = replyTo + } + + reqBody, _ := json.Marshal(body) + url := fmt.Sprintf("%s/open-apis/im/v1/messages?receive_id_type=%s", feishuAPIBase, receiverType) + req, _ := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(reqBody)) + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", "application/json") + + log.Printf("feishu: sending message to %s (%s): %s\n", receiveID, receiverType, text) + + resp, err := c.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + respBody, _ := io.ReadAll(resp.Body) + log.Printf("feishu: send response: status=%d, body=%s\n", resp.StatusCode, string(respBody)) + + var result struct { + Code int `json:"code"` + Msg string `json:"msg"` + Data *struct { + MessageID string `json:"message_id"` + } `json:"data"` + } + if err := json.Unmarshal(respBody, &result); err != nil { + return err + } + + if result.Code != 0 { + return fmt.Errorf("feishu API error: code=%d, msg=%s", result.Code, result.Msg) + } + + log.Printf("feishu: message sent successfully, message_id=%s\n", result.Data.MessageID) + return nil +} + +func (c *Channel) sendAttachment(ctx context.Context, chatID string, att bus.Attachment, msg bus.OutboundMessage) error { + token, err := c.getTenantAccessToken(ctx) + if err != nil { + return err + } + + var fileData []byte + var fileName string + + if att.Data != nil { + fileData = att.Data + fileName = att.Name + } else if att.LocalPath != "" { + data, err := os.ReadFile(att.LocalPath) + if err != nil { + return err + } + fileData = data + fileName = att.Name + } else { + return fmt.Errorf("attachment has no data or local path") + } + + fileID, err := c.uploadFile(ctx, token, fileData, fileName, att.MIMEType) + if err != nil { + return err + } + + msgType := "image" + contentKey := "image_key" + if strings.HasPrefix(att.MIMEType, "video/") { + msgType = "video" + contentKey = "video_key" + } else if !strings.HasPrefix(att.MIMEType, "image/") { + msgType = "file" + contentKey = "file_key" + } + + _, receiveID := c.resolveReceiverType(chatID) + + content := map[string]string{contentKey: fileID} + contentJSON, _ := json.Marshal(content) + + body := map[string]interface{}{ + "receive_id": receiveID, + "msg_type": msgType, + "content": string(contentJSON), + } + + replyTo := c.resolveReplyTo(msg) + if replyTo != "" { + body["reply_id"] = replyTo + } + + reqBody, _ := json.Marshal(body) + receiverType, _ := c.resolveReceiverType(chatID) + url := fmt.Sprintf("%s/open-apis/im/v1/messages?receive_id_type=%s", feishuAPIBase, receiverType) + req, _ := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(reqBody)) + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", "application/json") + + log.Printf("feishu: sending %s: key=%s\n", msgType, fileID) + + resp, err := c.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + + respBody, _ := io.ReadAll(resp.Body) + log.Printf("feishu: send response: status=%d, body=%s\n", resp.StatusCode, string(respBody)) + + var result struct { + Code int `json:"code"` + Msg string `json:"msg"` + Data *struct { + MessageID string `json:"message_id"` + } `json:"data"` + } + if err := json.Unmarshal(respBody, &result); err != nil { + return err + } + + if result.Code != 0 { + return fmt.Errorf("feishu API error: code=%d, msg=%s", result.Code, result.Msg) + } + + log.Printf("feishu: %s sent successfully, message_id=%s\n", msgType, result.Data.MessageID) + + return nil +} + +func (c *Channel) uploadFile(ctx context.Context, token string, data []byte, filename, mimeType string) (string, error) { + // Determine file type and use appropriate API + isImage := strings.HasPrefix(mimeType, "image/") + + var body bytes.Buffer + writer := multipart.NewWriter(&body) + + // Create form-data for file/image + fieldName := "file" + url := fmt.Sprintf("%s/open-apis/im/v1/files", feishuAPIBase) + + if isImage { + // Use image API for images + fieldName = "image" + url = fmt.Sprintf("%s/open-apis/im/v1/images", feishuAPIBase) + } + + filePart, err := writer.CreateFormFile(fieldName, filename) + if err != nil { + return "", err + } + if _, err := filePart.Write(data); err != nil { + return "", err + } + + if err := writer.Close(); err != nil { + return "", err + } + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, &body) + if err != nil { + return "", err + } + + req.Header.Set("Authorization", "Bearer "+token) + req.Header.Set("Content-Type", writer.FormDataContentType()) + + log.Printf("feishu: uploading %s (name=%s, size=%d) to %s\n", fieldName, filename, len(data), url) + + resp, err := c.client.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + + respBody, _ := io.ReadAll(resp.Body) + log.Printf("feishu: upload response: status=%d, body=%s\n", resp.StatusCode, string(respBody)) + + if isImage { + // Parse image response + var result struct { + Code int `json:"code"` + Msg string `json:"msg"` + Data struct { + ImageKey string `json:"image_key"` + } `json:"data"` + } + if err := json.Unmarshal(respBody, &result); err != nil { + return "", err + } + + if result.Code != 0 { + return "", fmt.Errorf("feishu upload error: code=%d, msg=%s", result.Code, result.Msg) + } + + if result.Data.ImageKey == "" { + return "", fmt.Errorf("feishu upload succeeded but no image_key returned") + } + log.Printf("feishu: image upload succeeded, image_key=%s\n", result.Data.ImageKey) + return result.Data.ImageKey, nil + } else { + // Parse file response + var result struct { + Code int `json:"code"` + Msg string `json:"msg"` + Data struct { + FileKey string `json:"file_key"` + FileToken string `json:"file_token"` + } `json:"data"` + } + if err := json.Unmarshal(respBody, &result); err != nil { + return "", err + } + + if result.Code != 0 { + return "", fmt.Errorf("feishu upload error: code=%d, msg=%s", result.Code, result.Msg) + } + + key := result.Data.FileKey + if key == "" { + key = result.Data.FileToken + } + if key == "" { + return "", fmt.Errorf("feishu upload succeeded but no file_key returned") + } + log.Printf("feishu: file upload succeeded, file_key=%s\n", key) + return key, nil + } +} + +func (c *Channel) getTenantAccessToken(ctx context.Context) (string, error) { + url := fmt.Sprintf("%s/open-apis/auth/v3/tenant_access_token/internal", feishuAPIBase) + body := map[string]string{ + "app_id": c.appID, + "app_secret": c.appSecret, + } + reqBody, _ := json.Marshal(body) + + req, _ := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(reqBody)) + req.Header.Set("Content-Type", "application/json") + + resp, err := c.client.Do(req) + if err != nil { + return "", err + } + defer resp.Body.Close() + + var result struct { + Code int `json:"code"` + Msg string `json:"msg"` + TenantAccessToken string `json:"tenant_access_token"` + Expire int `json:"expire"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return "", err + } + + if result.Code != 0 { + return "", fmt.Errorf("feishu auth error: code=%d, msg=%s", result.Code, result.Msg) + } + + return result.TenantAccessToken, nil +} + +func (c *Channel) resolveReceiverType(chatID string) (receiverType, receiveID string) { + chatID = strings.TrimSpace(chatID) + + if strings.HasPrefix(chatID, "ou_") { + return "open_id", chatID + } + if strings.HasPrefix(chatID, "u_") { + return "user_id", chatID + } + if strings.HasPrefix(chatID, "on_") { + return "union_id", chatID + } + return "chat_id", chatID +} + +func (c *Channel) resolveReplyTo(msg bus.OutboundMessage) string { + if strings.TrimSpace(msg.Delivery.ReplyToID) != "" { + return msg.Delivery.ReplyToID + } + if strings.TrimSpace(msg.ReplyTo) != "" { + return msg.ReplyTo + } + if strings.TrimSpace(msg.Delivery.ThreadID) != "" { + return msg.Delivery.ThreadID + } + return "" +} + +func min(a, b int) int { + if a < b { + return a + } + return b +} + +func randInt(min, max int) int { + return min + int(time.Now().UnixNano()%int64(max-min)) +} diff --git a/channels/feishu/feishu_test.go b/channels/feishu/feishu_test.go new file mode 100644 index 0000000..f9c4e1e --- /dev/null +++ b/channels/feishu/feishu_test.go @@ -0,0 +1,277 @@ +package feishu + +import ( + "context" + "testing" + + "github.com/mosaxiv/clawlet/bus" + "github.com/mosaxiv/clawlet/config" +) + +func TestFeishuChannelName(t *testing.T) { + b := bus.New(8) + cfg := config.FeishuConfig{ + Enabled: true, + AppID: "cli_test", + AppSecret: "test_secret", + } + ch := New(cfg, b) + + if ch.Name() != "feishu" { + t.Errorf("expected name 'feishu', got %s", ch.Name()) + } +} + +func TestFeishuChannelNotRunningInitially(t *testing.T) { + b := bus.New(8) + cfg := config.FeishuConfig{ + Enabled: true, + AppID: "cli_test", + AppSecret: "test_secret", + } + ch := New(cfg, b) + + if ch.IsRunning() { + t.Error("expected channel to not be running initially") + } +} + +func TestFeishuChannelRequiresAppID(t *testing.T) { + b := bus.New(8) + cfg := config.FeishuConfig{ + Enabled: true, + AppID: "", + AppSecret: "test_secret", + } + ch := New(cfg, b) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + err := ch.Start(ctx) + if err == nil { + t.Error("expected error for empty AppID") + } +} + +func TestFeishuChannelRequiresAppSecret(t *testing.T) { + b := bus.New(8) + cfg := config.FeishuConfig{ + Enabled: true, + AppID: "cli_test", + AppSecret: "", + } + ch := New(cfg, b) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + err := ch.Start(ctx) + if err == nil { + t.Error("expected error for empty AppSecret") + } +} + +func TestFeishuStripBotMention(t *testing.T) { + b := bus.New(8) + cfg := config.FeishuConfig{ + Enabled: true, + AppID: "cli_test", + AppSecret: "test_secret", + } + ch := New(cfg, b) + + tests := []struct { + name string + input string + expected string + }{ + { + name: "no mention", + input: "hello", + expected: "hello", + }, + { + name: "mention at start", + input: `bot hello`, + expected: "hello", + }, + { + name: "mention with colon", + input: `bot: hello`, + expected: "hello", + }, + { + name: "mention with comma", + input: `bot, hello`, + expected: "hello", + }, + { + name: "chinese punctuation", + input: `bot:你好`, + expected: "你好", + }, + { + name: "multiple mentions", + input: `botuser hello`, + expected: "hello", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result := ch.stripBotMention(tt.input) + if result != tt.expected { + t.Errorf("expected %q, got %q", tt.expected, result) + } + }) + } +} + +func TestFeishuResolveReceiverType(t *testing.T) { + b := bus.New(8) + cfg := config.FeishuConfig{ + Enabled: true, + AppID: "cli_test", + AppSecret: "test_secret", + } + ch := New(cfg, b) + + tests := []struct { + name string + chatID string + expectedType string + expectedID string + }{ + { + name: "open_id", + chatID: "ou_xxxxx", + expectedType: "open_id", + expectedID: "ou_xxxxx", + }, + { + name: "user_id", + chatID: "u_xxxxx", + expectedType: "user_id", + expectedID: "u_xxxxx", + }, + { + name: "union_id", + chatID: "on_xxxxx", + expectedType: "union_id", + expectedID: "on_xxxxx", + }, + { + name: "chat_id p2p", + chatID: "oc_xxxxx", + expectedType: "chat_id", + expectedID: "oc_xxxxx", + }, + { + name: "chat_id group", + chatID: "ch_xxxxx", + expectedType: "chat_id", + expectedID: "ch_xxxxx", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + receiverType, receiveID := ch.resolveReceiverType(tt.chatID) + if receiverType != tt.expectedType { + t.Errorf("expected receiver type %s, got %s", tt.expectedType, receiverType) + } + if receiveID != tt.expectedID { + t.Errorf("expected receiver ID %s, got %s", tt.expectedID, receiveID) + } + }) + } +} + +func TestFeishuResolveReplyTo(t *testing.T) { + b := bus.New(8) + cfg := config.FeishuConfig{ + Enabled: true, + AppID: "cli_test", + AppSecret: "test_secret", + } + ch := New(cfg, b) + + tests := []struct { + name string + msg bus.OutboundMessage + expectedReply string + }{ + { + name: "from Delivery.ReplyToID", + msg: bus.OutboundMessage{ + Delivery: bus.Delivery{ + ReplyToID: "msg123", + }, + }, + expectedReply: "msg123", + }, + { + name: "from ReplyTo field", + msg: bus.OutboundMessage{ + ReplyTo: "msg456", + }, + expectedReply: "msg456", + }, + { + name: "from Delivery.ThreadID", + msg: bus.OutboundMessage{ + Delivery: bus.Delivery{ + ThreadID: "thread789", + }, + }, + expectedReply: "thread789", + }, + { + name: "empty", + msg: bus.OutboundMessage{}, + expectedReply: "", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + result := ch.resolveReplyTo(tt.msg) + if result != tt.expectedReply { + t.Errorf("expected %s, got %s", tt.expectedReply, result) + } + }) + } +} + +func TestFeishuAllowList(t *testing.T) { + b := bus.New(8) + + // Test with empty allowFrom (allow all) + cfg := config.FeishuConfig{ + Enabled: true, + AppID: "cli_test", + AppSecret: "test_secret", + AllowFrom: []string{}, + } + ch := New(cfg, b) + + ctx := context.Background() + + // Create a test message + msg := &feishuEventMessage{ + MessageID: strPtr("msg123"), + ChatID: strPtr("oc_test"), + SenderID: &feishuUserID{OpenId: strPtr("ou_sender")}, + MsgType: strPtr("text"), + Content: strPtr(`{"text":"hello"}`), + ChatType: strPtr("p2p"), + } + + // Should not panic with empty allowFrom + ch.processMessageV2(ctx, msg, []byte{}, nil, "ou_sender") +} + +func strPtr(s string) *string { + return &s +} diff --git a/channels/slack/slack.go b/channels/slack/slack.go index 2dc3632..2364de7 100644 --- a/channels/slack/slack.go +++ b/channels/slack/slack.go @@ -1,9 +1,12 @@ package slack import ( + "bytes" "context" "fmt" + "io" "net/http" + "os" "strings" "sync" "sync/atomic" @@ -151,7 +154,8 @@ func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { return fmt.Errorf("chat_id is empty") } text := strings.TrimSpace(msg.Content) - if text == "" { + attachments := msg.Attachments + if text == "" && len(attachments) == 0 { return nil } c.mu.Lock() @@ -175,6 +179,8 @@ func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { c.mu.Unlock() } + // Send text message first if present + if text != "" { threadTS, direct := slackThreadMeta(msg) opts := []slack.MsgOption{ slack.MsgOptionText(text, false), @@ -184,6 +190,45 @@ func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { opts = append(opts, slack.MsgOptionTS(threadTS)) } _, _, err := api.PostMessageContext(ctx, ch, opts...) + if err != nil { + return err +} + } + + // Send attachments + for _, att := range attachments { + if err := c.sendSlackAttachment(ctx, api, ch, att, msg); err != nil { + return err + } + } + + return nil +} + +func (c *Channel) sendSlackAttachment(ctx context.Context, api *slack.Client, ch string, att bus.Attachment, msg bus.OutboundMessage) error { + var reader io.Reader + var filename string + + if att.Data != nil { + reader = bytes.NewReader(att.Data) + filename = att.Name + } else if att.LocalPath != "" { + file, err := os.Open(att.LocalPath) + if err != nil { + return err + } + defer file.Close() + reader = file + filename = att.Name + } else { + return fmt.Errorf("attachment has no data or local path") + } + + _, err := api.UploadFileContext(ctx, slack.FileUploadParameters{ + Reader: reader, + Filename: filename, + Channels: []string{ch}, + }) return err } diff --git a/channels/telegram/telegram.go b/channels/telegram/telegram.go index fffe233..4041c61 100644 --- a/channels/telegram/telegram.go +++ b/channels/telegram/telegram.go @@ -1,11 +1,13 @@ package telegram import ( + "bytes" "context" "errors" "fmt" "net" "net/http" + "os" "strconv" "strings" "sync" @@ -110,7 +112,8 @@ func (c *Channel) Stop() error { func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { text := strings.TrimSpace(msg.Content) - if text == "" { + attachments := msg.Attachments + if text == "" && len(attachments) == 0 { return nil } @@ -126,6 +129,8 @@ func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { return fmt.Errorf("telegram not connected") } + // Send text message if present + if text != "" { params := &tgbot.SendMessageParams{ ChatID: chatIDAny, Text: markdownToTelegramHTML(text), @@ -138,14 +143,114 @@ func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { } } if err := c.sendMessageWithRetry(ctx, b, params); err == nil { - return nil + // Continue to send attachments } else if !isTelegramParseError(err) { return err - } - + } else { params.Text = text params.ParseMode = "" - return c.sendMessageWithRetry(ctx, b, params) + if err := c.sendMessageWithRetry(ctx, b, params); err != nil { + return err +} + } + } + + // Send attachments + for _, att := range attachments { + if err := c.sendTelegramAttachment(ctx, b, chatIDAny, att, msg); err != nil { + return err + } + } + + return nil +} + +func (c *Channel) sendTelegramAttachment(ctx context.Context, b *tgbot.Bot, chatID any, att bus.Attachment, msg bus.OutboundMessage) error { + var inputFile models.InputFile + + if att.Data != nil { + // Use in-memory data + inputFile = &models.InputFileUpload{ + Filename: att.Name, + Data: bytes.NewReader(att.Data), + } + } else if att.LocalPath != "" { + // Use local file path + file, err := os.Open(att.LocalPath) + if err != nil { + return err + } + defer file.Close() + inputFile = &models.InputFileUpload{ + Filename: att.Name, + Data: file, + } + } else { + return fmt.Errorf("attachment has no data or local path") + } + + replyTo := resolveTelegramReplyTarget(msg) + + // Determine send method based on attachment kind or MIME type + switch strings.ToLower(att.Kind) { + case "image": + params := &tgbot.SendPhotoParams{ + ChatID: chatID, + Photo: inputFile, + Caption: att.Name, + } + if replyTo > 0 { + params.ReplyParameters = &models.ReplyParameters{ + MessageID: int(replyTo), + AllowSendingWithoutReply: true, + } + } + _, err := b.SendPhoto(ctx, params) + return err + case "audio": + params := &tgbot.SendAudioParams{ + ChatID: chatID, + Audio: inputFile, + Caption: att.Name, + } + if replyTo > 0 { + params.ReplyParameters = &models.ReplyParameters{ + MessageID: int(replyTo), + AllowSendingWithoutReply: true, + } + } + _, err := b.SendAudio(ctx, params) + return err + case "video": + params := &tgbot.SendVideoParams{ + ChatID: chatID, + Video: inputFile, + Caption: att.Name, + } + if replyTo > 0 { + params.ReplyParameters = &models.ReplyParameters{ + MessageID: int(replyTo), + AllowSendingWithoutReply: true, + } + } + _, err := b.SendVideo(ctx, params) + return err + default: + // Send as document for all other types + params := &tgbot.SendDocumentParams{ + ChatID: chatID, + Document: inputFile, + Caption: att.Name, + } + if replyTo > 0 { + params.ReplyParameters = &models.ReplyParameters{ + MessageID: int(replyTo), + AllowSendingWithoutReply: true, + } + } + _, err := b.SendDocument(ctx, params) + return err + } } func (c *Channel) onUpdate(ctx context.Context, b *tgbot.Bot, up *models.Update) { diff --git a/channels/whatsapp/whatsapp.go b/channels/whatsapp/whatsapp.go index e8f4bf2..732c23a 100644 --- a/channels/whatsapp/whatsapp.go +++ b/channels/whatsapp/whatsapp.go @@ -147,7 +147,8 @@ func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { return err } text := strings.TrimSpace(msg.Content) - if text == "" { + attachments := msg.Attachments + if text == "" && len(attachments) == 0 { return nil } @@ -158,11 +159,29 @@ func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { return fmt.Errorf("whatsapp not connected") } - payload := buildOutboundMessage(text, resolveWhatsAppReplyTarget(msg)) + // Send text message if present + if text != "" { + if err := c.sendWhatsAppMessage(ctx, wa, to, text); err != nil { + return err + } + } + + // Send attachments + for _, att := range attachments { + if err := c.sendWhatsAppAttachment(ctx, wa, to, att); err != nil { + return err + } + } + + return nil +} + +func (c *Channel) sendWhatsAppMessage(ctx context.Context, wa *whatsmeow.Client, to types.JID, text string) error { + payload := buildOutboundMessage(text, "") const maxAttempts = 3 for attempt := 1; attempt <= maxAttempts; attempt++ { - _, err = wa.SendMessage(ctx, to, payload) + _, err := wa.SendMessage(ctx, to, payload) if err == nil { return nil } @@ -181,6 +200,78 @@ func (c *Channel) Send(ctx context.Context, msg bus.OutboundMessage) error { return nil } +func (c *Channel) sendWhatsAppAttachment(ctx context.Context, wa *whatsmeow.Client, to types.JID, att bus.Attachment) error { + var data []byte + var err error + + if att.Data != nil { + data = att.Data + } else if att.LocalPath != "" { + data, err = os.ReadFile(att.LocalPath) + if err != nil { + return err + } + } else { + return fmt.Errorf("attachment has no data or local path") + } + + uploaded, err := wa.Upload(ctx, data, whatsmeow.MediaImage) // Default to image, will be overridden + if err != nil { + return err + } + + msg := &waE2E.Message{} + fileLength := uint64(len(data)) + switch strings.ToLower(att.Kind) { + case "image": + msg.ImageMessage = &waE2E.ImageMessage{ + Mimetype: &att.MIMEType, + Caption: &att.Name, + URL: &uploaded.URL, + DirectPath: &uploaded.DirectPath, + MediaKey: uploaded.MediaKey, + FileEncSHA256: uploaded.FileEncSHA256, + FileSHA256: uploaded.FileSHA256, + FileLength: &fileLength, + } + case "audio": + msg.AudioMessage = &waE2E.AudioMessage{ + Mimetype: &att.MIMEType, + URL: &uploaded.URL, + DirectPath: &uploaded.DirectPath, + MediaKey: uploaded.MediaKey, + FileEncSHA256: uploaded.FileEncSHA256, + FileSHA256: uploaded.FileSHA256, + FileLength: &fileLength, + } + case "video": + msg.VideoMessage = &waE2E.VideoMessage{ + Mimetype: &att.MIMEType, + Caption: &att.Name, + URL: &uploaded.URL, + DirectPath: &uploaded.DirectPath, + MediaKey: uploaded.MediaKey, + FileEncSHA256: uploaded.FileEncSHA256, + FileSHA256: uploaded.FileSHA256, + FileLength: &fileLength, + } + default: + msg.DocumentMessage = &waE2E.DocumentMessage{ + Mimetype: &att.MIMEType, + FileName: &att.Name, + URL: &uploaded.URL, + DirectPath: &uploaded.DirectPath, + MediaKey: uploaded.MediaKey, + FileEncSHA256: uploaded.FileEncSHA256, + FileSHA256: uploaded.FileSHA256, + FileLength: &fileLength, + } + } + + _, err = wa.SendMessage(ctx, to, msg) + return err +} + func (c *Channel) handleEvent(raw any) { switch evt := raw.(type) { case *events.Message: diff --git a/cmd/clawlet/cmd_channels.go b/cmd/clawlet/cmd_channels.go index b45e6ad..5239e06 100644 --- a/cmd/clawlet/cmd_channels.go +++ b/cmd/clawlet/cmd_channels.go @@ -31,6 +31,7 @@ func cmdChannels() *cli.Command { fmt.Printf("slack.enabled=%v\n", cfg.Channels.Slack.Enabled) fmt.Printf("telegram.enabled=%v\n", cfg.Channels.Telegram.Enabled) fmt.Printf("whatsapp.enabled=%v\n", cfg.Channels.WhatsApp.Enabled) + fmt.Printf("feishu.enabled=%v\n", cfg.Channels.Feishu.Enabled) return nil }, }, diff --git a/cmd/clawlet/cmd_gateway.go b/cmd/clawlet/cmd_gateway.go index 2147758..7e2570d 100644 --- a/cmd/clawlet/cmd_gateway.go +++ b/cmd/clawlet/cmd_gateway.go @@ -7,11 +7,13 @@ import ( "os" "os/signal" "strings" + "unicode/utf8" "github.com/mosaxiv/clawlet/agent" "github.com/mosaxiv/clawlet/bus" "github.com/mosaxiv/clawlet/channels" "github.com/mosaxiv/clawlet/channels/discord" + "github.com/mosaxiv/clawlet/channels/feishu" "github.com/mosaxiv/clawlet/channels/slack" "github.com/mosaxiv/clawlet/channels/telegram" "github.com/mosaxiv/clawlet/channels/whatsapp" @@ -60,7 +62,12 @@ func cmdGateway() *cli.Command { } ch := job.Payload.Channel to := job.Payload.To - if !job.Payload.Deliver || strings.TrimSpace(ch) == "" || strings.TrimSpace(to) == "" { + if !job.Payload.Deliver { + fmt.Printf("cron job %s executed (no delivery)\n", job.ID) + return "", nil + } + if strings.TrimSpace(ch) == "" || strings.TrimSpace(to) == "" { + fmt.Printf("cron job %s executed but no channel/to specified\n", job.ID) return "", nil } _ = b.PublishInbound(ctx, bus.InboundMessage{ @@ -98,12 +105,29 @@ func cmdGateway() *cli.Command { } } + onHeartbeat := func(ctx context.Context, prompt string) (string, error) { + resp, err := loop.ProcessDirect(ctx, prompt, "heartbeat", "cli", "heartbeat") + if err != nil { + return "", err + } + if strings.TrimSpace(cfg.Heartbeat.TargetChannel) != "" && strings.TrimSpace(cfg.Heartbeat.TargetChatID) != "" { + msg := "heartbeat OK" + if strings.TrimSpace(resp) != "" && !heartbeat.IsHeartbeatOK(resp) { + msg = "heartbeat FAILED: " + truncateForLog(resp, 400) + } + _ = b.PublishOutbound(ctx, bus.OutboundMessage{ + Channel: cfg.Heartbeat.TargetChannel, + ChatID: cfg.Heartbeat.TargetChatID, + Content: msg, + }) + } + return resp, nil + } + hb := heartbeat.New(wsAbs, heartbeat.Options{ Enabled: cfg.Heartbeat.EnabledValue(), IntervalSec: cfg.Heartbeat.IntervalSec, - OnHeartbeat: func(ctx context.Context, prompt string) (string, error) { - return loop.ProcessDirect(ctx, prompt, "heartbeat", "cli", "heartbeat") - }, + OnHeartbeat: onHeartbeat, }) hb.Start(ctx) @@ -138,6 +162,15 @@ func cmdGateway() *cli.Command { } cm.Add(whatsapp.New(cfg.Channels.WhatsApp, b)) } + if cfg.Channels.Feishu.Enabled { + if strings.TrimSpace(cfg.Channels.Feishu.AppID) == "" { + return fmt.Errorf("feishu enabled but appId is empty") + } + if strings.TrimSpace(cfg.Channels.Feishu.AppSecret) == "" { + return fmt.Errorf("feishu enabled but appSecret is empty") + } + cm.Add(feishu.New(cfg.Channels.Feishu, b)) + } if err := cm.StartAll(ctx); err != nil { return err @@ -197,3 +230,23 @@ func isLocalGatewayHost(host string) bool { } return false } + +func truncateForLog(s string, max int) string { + s = strings.TrimSpace(s) + if max <= 0 { + max = 200 + } + if len(s) <= max { + return s + } + // Truncate at UTF-8 boundary to avoid invalid encoding + truncated := s + for len(truncated) > max-3 { // Reserve space for "..." + _, size := utf8.DecodeLastRuneInString(truncated) + if size == 0 { + break + } + truncated = truncated[:len(truncated)-size] + } + return truncated + "...(truncated)" +} diff --git a/cmd/clawlet/cmd_sessions.go b/cmd/clawlet/cmd_sessions.go new file mode 100644 index 0000000..678fc46 --- /dev/null +++ b/cmd/clawlet/cmd_sessions.go @@ -0,0 +1,38 @@ +package main + +import ( + "context" + "fmt" + + "github.com/mosaxiv/clawlet/paths" + "github.com/mosaxiv/clawlet/session" + "github.com/urfave/cli/v3" +) + +func cmdSessions() *cli.Command { + return &cli.Command{ + Name: "sessions", + Usage: "inspect local session state", + Commands: []*cli.Command{ + { + Name: "list", + Usage: "list stored session keys", + Action: func(ctx context.Context, cmd *cli.Command) error { + dir := paths.SessionsDir() + keys, err := session.ListKeys(dir) + if err != nil { + return err + } + if len(keys) == 0 { + fmt.Printf("no sessions found in %s\n", dir) + return nil + } + for _, k := range keys { + fmt.Println(k) + } + return nil + }, + }, + }, + } +} diff --git a/cmd/clawlet/cmd_stats.go b/cmd/clawlet/cmd_stats.go new file mode 100644 index 0000000..914888b --- /dev/null +++ b/cmd/clawlet/cmd_stats.go @@ -0,0 +1,56 @@ +package main + +import ( + "context" + "fmt" + "sort" + "strings" + + "github.com/mosaxiv/clawlet/llm" + "github.com/urfave/cli/v3" +) + +func cmdStats() *cli.Command { + return &cli.Command{ + Name: "stats", + Usage: "print persisted statistics", + Commands: []*cli.Command{ + { + Name: "tokens", + Usage: "print token usage (per day)", + Flags: []cli.Flag{ + &cli.StringFlag{Name: "date", Usage: "YYYY-MM-DD (default: all dates)"}, + }, + Action: func(ctx context.Context, cmd *cli.Command) error { + date := strings.TrimSpace(cmd.String("date")) + store := llm.DefaultTokenUsageStore() + all := store.All() + if len(all) == 0 { + fmt.Println("no token usage recorded") + return nil + } + + keys := make([]string, 0, len(all)) + for k := range all { + if date != "" && k != date { + continue + } + keys = append(keys, k) + } + sort.Strings(keys) + if len(keys) == 0 { + fmt.Printf("no token usage recorded for %s\n", date) + return nil + } + + fmt.Printf("date calls prompt completion total\n") + for _, d := range keys { + u := all[d] + fmt.Printf("%s %5d %6d %10d %5d\n", d, u.Calls, u.PromptTokens, u.CompletionTokens, u.TotalTokens) + } + return nil + }, + }, + }, + } +} diff --git a/cmd/clawlet/cmd_status.go b/cmd/clawlet/cmd_status.go index 08a7cc9..19d8d59 100644 --- a/cmd/clawlet/cmd_status.go +++ b/cmd/clawlet/cmd_status.go @@ -28,6 +28,7 @@ func cmdStatus() *cli.Command { } fmt.Printf("agents.defaults.maxTokens: %d\n", cfg.Agents.Defaults.MaxTokensValue()) fmt.Printf("agents.defaults.temperature: %.2f\n", cfg.Agents.Defaults.TemperatureValue()) + fmt.Printf("log.level: %s\n", cfg.Log.LevelValue()) fmt.Printf("tools.restrictToWorkspace: %v\n", cfg.Tools.RestrictToWorkspaceValue()) fmt.Printf("tools.exec.timeoutSec: %d\n", cfg.Tools.Exec.TimeoutSec) fmt.Printf("tools.web.braveApiKey: %v\n", cfg.Tools.Web.BraveAPIKey != "") diff --git a/cmd/clawlet/main.go b/cmd/clawlet/main.go index e590495..9bf5311 100644 --- a/cmd/clawlet/main.go +++ b/cmd/clawlet/main.go @@ -3,12 +3,30 @@ package main import ( "context" "fmt" + "log" "os" + "path/filepath" + "github.com/mosaxiv/clawlet/paths" "github.com/urfave/cli/v3" ) func main() { + // Get config directory for log file + configDir, err := paths.ConfigDir() + if err != nil { + log.Fatal("Failed to get config directory:", err) + } + logPath := filepath.Join(configDir, "clawlet.log") + + file, err := os.OpenFile(logPath, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644) + if err != nil { + log.Fatal("Failed to open log file:", err) + } + defer file.Close() + log.SetOutput(file) + log.SetFlags(log.Ldate | log.Ltime | log.Lshortfile) + root := &cli.Command{ Name: "clawlet", Usage: "minimal Go agent", @@ -22,6 +40,8 @@ func main() { cmdProvider(), cmdChannels(), cmdCron(), + cmdStats(), + cmdSessions(), }, } diff --git a/config/config.go b/config/config.go index 33640a3..ea9aa24 100644 --- a/config/config.go +++ b/config/config.go @@ -18,6 +18,7 @@ type Config struct { Cron CronConfig `json:"cron"` Heartbeat HeartbeatConfig `json:"heartbeat"` Gateway GatewayConfig `json:"gateway"` + Log LogConfig `json:"log,omitempty"` // Channels are optional; enable what you need. Channels ChannelsConfig `json:"channels"` } @@ -238,6 +239,17 @@ func (c MediaToolsConfig) AttachmentEnabledValue() bool { return *c.AttachmentEnabled } +type LogConfig struct { + Level string `json:"level,omitempty"` +} + +func (c LogConfig) LevelValue() string { + if strings.TrimSpace(c.Level) == "" { + return "warning" + } + return strings.ToLower(strings.TrimSpace(c.Level)) +} + type CronConfig struct { Enabled *bool `json:"enabled"` } @@ -250,8 +262,10 @@ func (c CronConfig) EnabledValue() bool { } type HeartbeatConfig struct { - Enabled *bool `json:"enabled"` - IntervalSec int `json:"intervalSec"` + Enabled *bool `json:"enabled"` + IntervalSec int `json:"intervalSec"` + TargetChannel string `json:"targetChannel,omitempty"` // e.g. "telegram" + TargetChatID string `json:"targetChatId,omitempty"` } func (c HeartbeatConfig) EnabledValue() bool { @@ -275,6 +289,7 @@ type ChannelsConfig struct { Slack SlackConfig `json:"slack"` Telegram TelegramConfig `json:"telegram"` WhatsApp WhatsAppConfig `json:"whatsapp"` + Feishu FeishuConfig `json:"feishu"` } type DiscordConfig struct { @@ -320,6 +335,14 @@ type WhatsAppConfig struct { SessionStorePath string `json:"sessionStorePath,omitempty"` // optional: sqlite store path for persistent login } +// Feishu/Lark (WebSocket long connection, no public IP needed). +type FeishuConfig struct { + Enabled bool `json:"enabled"` + AllowFrom []string `json:"allowFrom"` + AppID string `json:"appId"` // App ID from Feishu developer console + AppSecret string `json:"appSecret"` // App Secret from Feishu developer console +} + const ( DefaultAgentMaxTokens = 8192 DefaultAgentTemperature = 0.7 @@ -496,6 +519,12 @@ func Default() *Config { Enabled: false, AllowFrom: nil, }, + Feishu: FeishuConfig{ + Enabled: false, + AllowFrom: nil, + AppID: "", + AppSecret: "", + }, }, } } diff --git a/go.mod b/go.mod index 5b6eab4..b1408c6 100644 --- a/go.mod +++ b/go.mod @@ -5,7 +5,9 @@ go 1.26.0 require ( github.com/bwmarrin/discordgo v0.29.0 github.com/go-telegram/bot v1.19.0 + github.com/gorilla/websocket v1.5.3 github.com/hashicorp/go-retryablehttp v0.7.8 + github.com/larksuite/oapi-sdk-go/v3 v3.5.3 github.com/mdp/qrterminal/v3 v3.2.1 github.com/ncruces/go-sqlite3 v0.30.5 github.com/slack-go/slack v0.17.3 @@ -19,8 +21,8 @@ require ( github.com/beeper/argo-go v1.1.2 // indirect github.com/coder/websocket v1.8.14 // indirect github.com/elliotchance/orderedmap/v3 v3.1.0 // indirect + github.com/gogo/protobuf v1.3.2 // indirect github.com/google/uuid v1.6.0 // indirect - github.com/gorilla/websocket v1.5.3 // indirect github.com/hashicorp/go-cleanhttp v0.5.2 // indirect github.com/mattn/go-colorable v0.1.14 // indirect github.com/mattn/go-isatty v0.0.20 // indirect diff --git a/go.sum b/go.sum index 920763f..d025f95 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,3 @@ -filippo.io/edwards25519 v1.1.0 h1:FNf4tywRC1HmFuKW5xopWpigGjJKiJSV0Cqo0cJWDaA= -filippo.io/edwards25519 v1.1.0/go.mod h1:BxyFTGdWcka3PhytdK4V28tE5sGfRvvvRV7EaN4VDT4= filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo= filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc= github.com/DATA-DOG/go-sqlmock v1.5.2 h1:OcvFkGmslmlZibjAjaHm3L//6LiuBgolP7OputlJIzU= @@ -21,18 +19,19 @@ github.com/elliotchance/orderedmap/v3 v3.1.0 h1:j4DJ5ObEmMBt/lcwIecKcoRxIQUEnw0L github.com/elliotchance/orderedmap/v3 v3.1.0/go.mod h1:G+Hc2RwaZvJMcS4JpGCOyViCnGeKf0bTYCGTO4uhjSo= github.com/fatih/color v1.16.0 h1:zmkK9Ngbjj+K0yRhTVONQh1p/HknKYSlNT+vZCzyokM= github.com/fatih/color v1.16.0/go.mod h1:fL2Sau1YI5c0pdGEVCbKQbLXB6edEj1ZgiY4NijnWvE= -github.com/go-telegram/bot v1.18.0 h1:yQzv437DY42SYTPBY48RinAvwbmf1ox5QICskIYWCD8= -github.com/go-telegram/bot v1.18.0/go.mod h1:i2TRs7fXWIeaceF3z7KzsMt/he0TwkVC680mvdTFYeM= github.com/go-telegram/bot v1.19.0 h1:tuvTQhgNietHFRN0HUDhuXsgfgkGSaO8WWwZQW3DMQg= github.com/go-telegram/bot v1.19.0/go.mod h1:i2TRs7fXWIeaceF3z7KzsMt/he0TwkVC680mvdTFYeM= github.com/go-test/deep v1.1.1 h1:0r/53hagsehfO4bzD2Pgr/+RgHqhmf+k1Bpse2cTu1U= github.com/go-test/deep v1.1.1/go.mod h1:5C2ZWiW0ErCdrYzpqxLbTX7MG14M9iiw8DgHncVwcsE= github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA= +github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= +github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/hashicorp/go-cleanhttp v0.5.2 h1:035FKYIWjmULyFRBKPs8TBQoi0x6d9G4xc9neXJWAZQ= @@ -41,6 +40,10 @@ github.com/hashicorp/go-hclog v1.6.3 h1:Qr2kF+eVWjTiYmU7Y31tYlP1h0q/X3Nl3tPGdaB1 github.com/hashicorp/go-hclog v1.6.3/go.mod h1:W4Qnvbt70Wk/zYJryRzDRU/4r0kIg0PVHBcfoyhpF5M= github.com/hashicorp/go-retryablehttp v0.7.8 h1:ylXZWnqa7Lhqpk0L1P1LzDtGcCR0rPVUrx/c8Unxc48= github.com/hashicorp/go-retryablehttp v0.7.8/go.mod h1:rjiScheydd+CxvumBsIrFKlx3iS0jrZ7LvzFGFmuKbw= +github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= +github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= +github.com/larksuite/oapi-sdk-go/v3 v3.5.3 h1:xvf8Dv29kBXC5/DNDCLhHkAFW8l/0LlQJimO5Zn+JUk= +github.com/larksuite/oapi-sdk-go/v3 v3.5.3/go.mod h1:ZEplY+kwuIrj/nqw5uSCINNATcH3KdxSN7y+UxYY5fI= github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg= github.com/mattn/go-colorable v0.1.14 h1:9A9LHSqF/7dyVVX6g0U9cwm9pG3kP9gSzcuIPHPsaIE= github.com/mattn/go-colorable v0.1.14/go.mod h1:6LmQG8QLFO4G5z1gPvYEzlUgJ2wF+stgPZH1UqBm1s8= @@ -48,9 +51,8 @@ github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/ github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY= github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= -github.com/mattn/go-sqlite3 v1.14.33 h1:A5blZ5ulQo2AtayQ9/limgHEkFreKj1Dv226a1K73s0= -github.com/mattn/go-sqlite3 v1.14.33/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y= github.com/mattn/go-sqlite3 v1.14.34 h1:3NtcvcUnFBPsuRcno8pUtupspG/GM+9nZ88zgJcp6Zk= +github.com/mattn/go-sqlite3 v1.14.34/go.mod h1:Uh1q+B4BYcTPb+yiD3kU8Ct7aC0hY9fxUwlHK0RXw+Y= github.com/mdp/qrterminal/v3 v3.2.1 h1:6+yQjiiOsSuXT5n9/m60E54vdgFsw0zhADHhHLrFet4= github.com/mdp/qrterminal/v3 v3.2.1/go.mod h1:jOTmXvnBsMy5xqLniO0R++Jmjs2sTm9dFSuQ5kpz/SU= github.com/ncruces/go-sqlite3 v0.30.5 h1:6usmTQ6khriL8oWilkAZSJM/AIpAlVL2zFrlcpDldCE= @@ -77,24 +79,37 @@ github.com/urfave/cli/v3 v3.6.2 h1:lQuqiPrZ1cIz8hz+HcrG0TNZFxU70dPZ3Yl+pSrH9A8= github.com/urfave/cli/v3 v3.6.2/go.mod h1:ysVLtOEmg2tOy6PknnYVhDoouyC/6N42TMeoMzskhso= github.com/vektah/gqlparser/v2 v2.5.31 h1:YhWGA1mfTjID7qJhd1+Vxhpk5HTgydrGU9IgkWBTJ7k= github.com/vektah/gqlparser/v2 v2.5.31/go.mod h1:c1I28gSOVNzlfc4WuDlqU7voQnsqI6OG2amkBAFmgts= +github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= go.mau.fi/libsignal v0.2.1 h1:vRZG4EzTn70XY6Oh/pVKrQGuMHBkAWlGRC22/85m9L0= go.mau.fi/libsignal v0.2.1/go.mod h1:iVvjrHyfQqWajOUaMEsIfo3IqgVMrhWcPiiEzk7NgoU= -go.mau.fi/util v0.9.5 h1:7AoWPCIZJGv4jvtFEuCe3GhAbI7uF9ckIooaXvwlIR4= -go.mau.fi/util v0.9.5/go.mod h1:g1uvZ03VQhtTt2BgaRGVytS/Zj67NV0YNIECch0sQCQ= go.mau.fi/util v0.9.6 h1:2nsvxm49KhI3wrFltr0+wSUBlnQ4CMtykuELjpIU+ts= go.mau.fi/util v0.9.6/go.mod h1:sIJpRH7Iy5Ad1SBuxQoatxtIeErgzxCtjd/2hCMkYMI= -go.mau.fi/whatsmeow v0.0.0-20260211193157-7b33f6289f98 h1:4ePal8sykeD3vUcUWvECtfqoGyNr5UHYn8pPwrBittY= -go.mau.fi/whatsmeow v0.0.0-20260211193157-7b33f6289f98/go.mod h1:jDLOQLLiYXcm4vMB6vtPcBLU387sRY+P3vOElxX8srA= go.mau.fi/whatsmeow v0.0.0-20260218135554-9cbe80fb25a4 h1:+3FE6cq5NzELYVD7uxa0yDpbUB+poSQmJV8zENTjHZA= go.mau.fi/whatsmeow v0.0.0-20260218135554-9cbe80fb25a4/go.mod h1:mXCRFyPEPn4jqWz6Afirn8vY7DpHCPnlKq6I2cWwFHM= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= +golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= golang.org/x/crypto v0.0.0-20210421170649-83a5a9bb288b/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4= golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts= golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos= golang.org/x/exp v0.0.0-20260212183809-81e46e3db34a h1:ovFr6Z0MNmU7nH8VaX5xqw+05ST2uO1exVfZPVqRC5o= golang.org/x/exp v0.0.0-20260212183809-81e46e3db34a/go.mod h1:K79w1Vqn7PoiZn+TkNpx3BUWUQksGO3JcVX6qIjytmA= +golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg= golang.org/x/net v0.50.0 h1:ucWh9eiCGyDR3vtzso0WMQinm2Dnt8cFMuQa9K33J60= golang.org/x/net v0.50.0/go.mod h1:UgoSli3F/pBgdJBHCTc+tp3gmrU4XswgGRgtnwWTfyM= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= @@ -104,10 +119,18 @@ golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.40.0 h1:36e4zGLqU4yhjlmxEaagx2KuYbJq3EwY8K943ZsHcvg= golang.org/x/term v0.40.0/go.mod h1:w2P8uVp06p2iyKKuvXIm7N/y0UCRt3UfJTfZ7oOpglM= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk= golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= +golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= diff --git a/heartbeat/service.go b/heartbeat/service.go index ab6fef1..ac5a6f3 100644 --- a/heartbeat/service.go +++ b/heartbeat/service.go @@ -8,6 +8,7 @@ import ( "strings" "sync/atomic" "time" + "unicode/utf8" ) const ( @@ -43,6 +44,8 @@ func New(workspace string, opts Options) *Service { if sec <= 0 { sec = DefaultIntervalSec } + log.Printf("heartbeat: initialized with interval %v", sec) + return &Service{ workspace: workspace, onBeat: opts.OnHeartbeat, @@ -110,7 +113,7 @@ func (s *Service) tick(ctx context.Context) { log.Printf("heartbeat: error: %v", err) return } - if isHeartbeatOK(resp) { + if IsHeartbeatOK(resp) { return } if strings.TrimSpace(resp) != "" { @@ -157,8 +160,9 @@ func isEmpty(content string) bool { return true } -func isHeartbeatOK(resp string) bool { - // Tolerate case/underscore/whitespace variations. +// IsHeartbeatOK reports whether the response should be treated as an OK acknowledgement. +// It tolerates case / whitespace / underscore variations. +func IsHeartbeatOK(resp string) bool { return strings.Contains(normalizeToken(resp), normalizeToken(OKToken)) } @@ -170,7 +174,16 @@ func truncateForLog(s string, max int) string { if len(s) <= max { return s } - return s[:max] + "...(truncated)" + // Truncate at UTF-8 boundary to avoid invalid encoding + truncated := s + for len(truncated) > max-3 { // Reserve space for "..." + _, size := utf8.DecodeLastRuneInString(truncated) + if size == 0 { + break +} + truncated = truncated[:len(truncated)-size] + } + return truncated + "...(truncated)" } func normalizeToken(s string) string { diff --git a/heartbeat/service_test.go b/heartbeat/service_test.go index f42d4b8..291b8a3 100644 --- a/heartbeat/service_test.go +++ b/heartbeat/service_test.go @@ -17,8 +17,8 @@ func TestIsHeartbeatOK(t *testing.T) { {"nothing to do: HEARTBEAT_OK", true}, } for _, c := range cases { - if got := isHeartbeatOK(c.in); got != c.want { - t.Fatalf("isHeartbeatOK(%q)=%v want %v", c.in, got, c.want) + if got := IsHeartbeatOK(c.in); got != c.want { + t.Fatalf("IsHeartbeatOK(%q)=%v want %v", c.in, got, c.want) } } } diff --git a/llm/anthropic.go b/llm/anthropic.go index 630ea68..30c8e1c 100644 --- a/llm/anthropic.go +++ b/llm/anthropic.go @@ -58,12 +58,19 @@ func (c *Client) chatAnthropic(ctx context.Context, messages []Message, tools [] req.Header.Set(k, v) } + // Debug log request + c.debugLog("LLM Request to %s: %s", endpoint, string(b)) + resp, err := c.HTTP.Do(req) if err != nil { return nil, err } defer resp.Body.Close() body, _ := io.ReadAll(io.LimitReader(resp.Body, 8<<20)) + + // Debug log response + c.debugLog("LLM Response from %s (status %d): %s", endpoint, resp.StatusCode, string(body)) + if resp.StatusCode < 200 || resp.StatusCode >= 300 { return nil, fmt.Errorf("llm http %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) } @@ -109,6 +116,8 @@ func (c *Client) chatAnthropic(ctx context.Context, messages []Message, tools [] } } out.Content = strings.Join(textParts, "\n") + // Record token usage if available. + RecordTokenUsageFromResponse(body) return out, nil } diff --git a/llm/client.go b/llm/client.go index 3561ed9..a66ef16 100644 --- a/llm/client.go +++ b/llm/client.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "log" "net/http" "strings" "time" @@ -17,6 +18,7 @@ type Client struct { MaxTokens int Temperature *float64 Headers map[string]string + LogLevel string HTTP HTTPDoer } @@ -41,8 +43,9 @@ func (c *Client) Chat(ctx context.Context, messages []Message, tools []ToolDefin if c.HTTP == nil { c.HTTP = &http.Client{Timeout: 120 * time.Second} } + switch normalizeProvider(c.Provider) { - case "", "openai", "openrouter", "ollama","shengsuanyun": + case "", "openai", "openrouter", "ollama", "shengsuanyun": return c.chatOpenAICompatible(ctx, messages, tools) case "anthropic": return c.chatAnthropic(ctx, messages, tools) @@ -55,6 +58,22 @@ func (c *Client) Chat(ctx context.Context, messages []Message, tools []ToolDefin } } +func (c *Client) shouldLogDebug() bool { + level := strings.ToLower(strings.TrimSpace(c.LogLevel)) + switch level { + case "debug", "trace": + return true + default: + return false + } +} + +func (c *Client) debugLog(format string, args ...interface{}) { + if c.shouldLogDebug() { + log.Printf("[DEBUG] "+format, args...) + } +} + func normalizeProvider(p string) string { switch strings.ToLower(strings.TrimSpace(p)) { case "local": diff --git a/llm/gemini.go b/llm/gemini.go index ea92ee6..a2ee8ee 100644 --- a/llm/gemini.go +++ b/llm/gemini.go @@ -60,12 +60,19 @@ func (c *Client) chatGemini(ctx context.Context, messages []Message, tools []Too req.Header.Set(k, v) } + // Debug log request + c.debugLog("LLM Request to %s: %s", endpoint, string(b)) + resp, err := c.HTTP.Do(req) if err != nil { return nil, err } defer resp.Body.Close() body, _ := io.ReadAll(io.LimitReader(resp.Body, 8<<20)) + + // Debug log response + c.debugLog("LLM Response from %s (status %d): %s", endpoint, resp.StatusCode, string(body)) + if resp.StatusCode < 200 || resp.StatusCode >= 300 { return nil, fmt.Errorf("llm http %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) } @@ -117,6 +124,8 @@ func (c *Client) chatGemini(ctx context.Context, messages []Message, tools []Too } } out.Content = strings.Join(textParts, "\n") + // Record token usage if available. + RecordTokenUsageFromResponse(body) return out, nil } diff --git a/llm/openai.go b/llm/openai.go index f6a72c0..abf0792 100644 --- a/llm/openai.go +++ b/llm/openai.go @@ -51,6 +51,9 @@ func (c *Client) chatOpenAICompatible(ctx context.Context, messages []Message, t req.Header.Set(k, v) } + // Debug log request + c.debugLog("LLM Request to %s: %s", endpoint, string(b)) + hc := c.HTTP if hc == nil { hc = &http.Client{Timeout: 120 * time.Second} @@ -61,6 +64,10 @@ func (c *Client) chatOpenAICompatible(ctx context.Context, messages []Message, t } defer resp.Body.Close() body, _ := io.ReadAll(io.LimitReader(resp.Body, 8<<20)) + + // Debug log response + c.debugLog("LLM Response from %s (status %d): %s", endpoint, resp.StatusCode, string(body)) + if resp.StatusCode < 200 || resp.StatusCode >= 300 { return nil, fmt.Errorf("llm http %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) } @@ -104,6 +111,8 @@ func (c *Client) chatOpenAICompatible(ctx context.Context, messages []Message, t Arguments: args, }) } + // Record token usage if available. + RecordTokenUsageFromResponse(body) return out, nil } diff --git a/llm/token_usage.go b/llm/token_usage.go new file mode 100644 index 0000000..24dca35 --- /dev/null +++ b/llm/token_usage.go @@ -0,0 +1,215 @@ +package llm + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "sync" + "time" + + "github.com/mosaxiv/clawlet/paths" +) + +// DailyTokenUsage tracks token usage per calendar day. +type DailyTokenUsage struct { + Calls int `json:"calls"` + PromptTokens int `json:"prompt_tokens"` + CompletionTokens int `json:"completion_tokens"` + TotalTokens int `json:"total_tokens"` +} + +// TokenUsageStore is a persistent, thread-safe token usage tracker. +type TokenUsageStore struct { + mu sync.Mutex + path string + data map[string]*DailyTokenUsage +} + +// DefaultTokenUsageStore returns the singleton store used by the llm package. +func DefaultTokenUsageStore() *TokenUsageStore { + return globalTokenUsage +} + +// NewTokenUsageStore creates a usage store backed by the given file path. +// The file will be loaded if it exists. +func NewTokenUsageStore(path string) (*TokenUsageStore, error) { + store := &TokenUsageStore{ + path: path, + data: map[string]*DailyTokenUsage{}, + } + if err := store.load(); err != nil { + return nil, err + } + return store, nil +} + +func (s *TokenUsageStore) load() error { + if s.path == "" { + return fmt.Errorf("token usage path is empty") + } + b, err := os.ReadFile(s.path) + if err != nil { + if os.IsNotExist(err) { + return nil + } + return err + } + var raw map[string]DailyTokenUsage + if err := json.Unmarshal(b, &raw); err != nil { + return err + } + for k, v := range raw { + copy := v + s.data[k] = © + } + return nil +} + +func (s *TokenUsageStore) save() error { + if s.path == "" { + return fmt.Errorf("token usage path is empty") + } + if err := os.MkdirAll(filepath.Dir(s.path), 0o700); err != nil { + return err + } + out := map[string]DailyTokenUsage{} + for k, v := range s.data { + out[k] = *v + } + b, err := json.MarshalIndent(out, "", " ") + if err != nil { + return err + } + return os.WriteFile(s.path, b, 0o600) +} + +func (s *TokenUsageStore) Add(prompt, completion, total int) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.data == nil { + s.data = map[string]*DailyTokenUsage{} + } + key := time.Now().UTC().Format("2006-01-02") + d, ok := s.data[key] + if !ok { + d = &DailyTokenUsage{} + s.data[key] = d + } + d.Calls++ + d.PromptTokens += prompt + d.CompletionTokens += completion + d.TotalTokens += total + return s.save() +} + +func (s *TokenUsageStore) Get(day string) (DailyTokenUsage, bool) { + s.mu.Lock() + defer s.mu.Unlock() + if s.data == nil { + return DailyTokenUsage{}, false + } + d, ok := s.data[day] + if !ok { + return DailyTokenUsage{}, false + } + return *d, true +} + +func (s *TokenUsageStore) All() map[string]DailyTokenUsage { + s.mu.Lock() + defer s.mu.Unlock() + out := map[string]DailyTokenUsage{} + for k, v := range s.data { + out[k] = *v + } + return out +} + +var globalTokenUsage *TokenUsageStore + +func init() { + path := paths.TokenUsagePath() + store, err := NewTokenUsageStore(path) + if err != nil { + // Fail silently; token usage is non-critical. + store = &TokenUsageStore{path: path, data: map[string]*DailyTokenUsage{}} + } + globalTokenUsage = store +} + +func RecordTokenUsage(prompt, completion, total int) { + if globalTokenUsage == nil { + return + } + _ = globalTokenUsage.Add(prompt, completion, total) +} + +func RecordTokenUsageFromResponse(body []byte) { + prompt, completion, total := parseTokenUsage(body) + if prompt == 0 && completion == 0 && total == 0 { + return + } + RecordTokenUsage(prompt, completion, total) +} + +func parseTokenUsage(body []byte) (prompt, completion, total int) { + var m map[string]any + if err := json.Unmarshal(body, &m); err != nil { + return + } + + getInt := func(v any) int { + switch t := v.(type) { + case float64: + return int(t) + case int: + return t + case int64: + return int(t) + case json.Number: + i, _ := t.Int64() + return int(i) + default: + return 0 + } + } + + // OpenAI compatible + if u, ok := m["usage"].(map[string]any); ok { + prompt = getInt(u["prompt_tokens"]) + completion = getInt(u["completion_tokens"]) + total = getInt(u["total_tokens"]) + if total == 0 { + total = prompt + completion + } + } + + // Gemini tokenUsage + if total == 0 { + if u, ok := m["tokenUsage"].(map[string]any); ok { + prompt = getInt(u["promptTokens"]) + completion = getInt(u["completionTokens"]) + total = getInt(u["totalTokens"]) + if total == 0 { + total = prompt + completion + } + } + } + + // Anthropic / other providers might expose prompt/completion tokens at top level. + if prompt == 0 { + prompt = getInt(m["prompt_tokens"]) + } + if completion == 0 { + completion = getInt(m["completion_tokens"]) + } + if total == 0 { + total = getInt(m["total_tokens"]) + } + + if total == 0 { + total = prompt + completion + } + return +} diff --git a/llm/token_usage_test.go b/llm/token_usage_test.go new file mode 100644 index 0000000..7c93444 --- /dev/null +++ b/llm/token_usage_test.go @@ -0,0 +1,46 @@ +package llm + +import ( + "os" + "path/filepath" + "testing" + "time" +) + +func TestTokenUsageStore_AddAndPersist(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "usage.json") + + store, err := NewTokenUsageStore(path) + if err != nil { + t.Fatalf("NewTokenUsageStore: %v", err) + } + + // Add usage for today. + if err := store.Add(10, 20, 30); err != nil { + t.Fatalf("Add: %v", err) + } + + // Ensure file was written. + b, err := os.ReadFile(path) + if err != nil { + t.Fatalf("ReadFile: %v", err) + } + if len(b) == 0 { + t.Fatalf("expected non-empty file") + } + + // Reload and check values. + store2, err := NewTokenUsageStore(path) + if err != nil { + t.Fatalf("NewTokenUsageStore (reload): %v", err) + } + day := time.Now().UTC().Format("2006-01-02") + usage, ok := store2.Get(day) + if !ok { + t.Fatalf("expected usage for %q", day) + } + if usage.Calls != 1 || usage.PromptTokens != 10 || usage.CompletionTokens != 20 || usage.TotalTokens != 30 { + t.Fatalf("unexpected usage: %+v", usage) + } +} diff --git a/paths/paths.go b/paths/paths.go index 3dcc1bf..b4d9ff7 100644 --- a/paths/paths.go +++ b/paths/paths.go @@ -47,6 +47,14 @@ func WorkspaceDir() string { return filepath.Join(dir, "workspace") } +func TokenUsagePath() string { + dir, err := ConfigDir() + if err != nil { + return ".clawlet/token_usage.json" + } + return filepath.Join(dir, "token_usage.json") +} + func EnsureStateDirs() error { cfgDir, err := ConfigDir() if err != nil { diff --git a/session/session.go b/session/session.go index f3512f0..8d8e7b0 100644 --- a/session/session.go +++ b/session/session.go @@ -235,6 +235,12 @@ func Save(dir string, s *Session) error { s.mu.Lock() defer s.mu.Unlock() + // Ensure the session key is stored on disk so it can be recovered from the session file later. + if s.Metadata == nil { + s.Metadata = map[string]any{} + } + s.Metadata["session_key"] = s.Key + f, err := os.OpenFile(path, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600) if err != nil { return err @@ -272,6 +278,59 @@ func Save(dir string, s *Session) error { return nil } +// ListKeys returns the session keys for all sessions stored in the given directory. +// +// It reads the first JSON line of each session file and tries to extract the stored +// "session_key" metadata value. If the metadata is missing, it falls back to the +// filename (without the .jsonl extension). +func ListKeys(dir string) ([]string, error) { + files, err := os.ReadDir(dir) + if err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, err + } + var keys []string + for _, de := range files { + if de.IsDir() { + continue + } + name := de.Name() + if !strings.HasSuffix(name, ".jsonl") { + continue + } + path := filepath.Join(dir, name) + f, err := os.Open(path) + if err != nil { + continue + } + sc := bufio.NewScanner(f) + var k string + for sc.Scan() { + line := strings.TrimSpace(sc.Text()) + if line == "" { + continue + } + var ml metadataLine + if err := json.Unmarshal([]byte(line), &ml); err == nil { + if v, ok := ml.Metadata["session_key"]; ok { + if ks, ok := v.(string); ok && ks != "" { + k = ks + } + } + } + break + } + _ = f.Close() + if k == "" { + k = strings.TrimSuffix(name, ".jsonl") + } + keys = append(keys, k) + } + return keys, nil +} + func cloneMessages(in []Message) []Message { out := make([]Message, 0, len(in)) for _, m := range in { diff --git a/session/session_test.go b/session/session_test.go index d1a10b7..ae27382 100644 --- a/session/session_test.go +++ b/session/session_test.go @@ -90,3 +90,23 @@ func TestSave_AfterConsolidationPersistsTrimmedMessages(t *testing.T) { t.Fatalf("messages=%d want=%d", got, keep) } } + +func TestListKeysIncludesSavedSessionKey(t *testing.T) { + dir := t.TempDir() + key := "cli:test" + s := New(key) + if err := Save(dir, s); err != nil { + t.Fatalf("save: %v", err) + } + + keys, err := ListKeys(dir) + if err != nil { + t.Fatalf("list keys: %v", err) + } + if len(keys) != 1 { + t.Fatalf("keys=%v", keys) + } + if keys[0] != key { + t.Fatalf("unexpected key %q", keys[0]) + } +} diff --git a/tools/defs.go b/tools/defs.go index 16c6e48..0f216d2 100644 --- a/tools/defs.go +++ b/tools/defs.go @@ -213,6 +213,27 @@ func defMessage() llm.ToolDefinition { } } +func defSendFile() llm.ToolDefinition { + return llm.ToolDefinition{ + Type: "function", + Function: llm.FunctionDefinition{ + Name: "send_file", + Description: "Send a file to a specific channel/chat_id. Supports local files and data.", + Parameters: llm.JSONSchema{ + Type: "object", + Properties: map[string]llm.JSONSchema{ + "file_path": {Type: "string", Description: "Path to local file to send (relative to workspace)."}, + "file_name": {Type: "string", Description: "Display name for the file."}, + "channel": {Type: "string", Description: "Target channel (e.g., 'telegram')."}, + "chat_id": {Type: "string", Description: "Target chat ID."}, + "caption": {Type: "string", Description: "Optional caption/text to send with the file."}, + }, + Required: []string{"file_path", "channel", "chat_id"}, + }, + }, + } +} + func defSpawn() llm.ToolDefinition { return llm.ToolDefinition{ Type: "function", diff --git a/tools/registry.go b/tools/registry.go index 547369c..62aa330 100644 --- a/tools/registry.go +++ b/tools/registry.go @@ -65,7 +65,7 @@ func (r *Registry) Definitions() []llm.ToolDefinition { defs = append(defs, defWebSearch()) } if r.Outbound != nil { - defs = append(defs, defMessage()) + defs = append(defs, defMessage(), defSendFile()) } if r.Spawn != nil { defs = append(defs, defSpawn()) @@ -224,10 +224,34 @@ func (r *Registry) Execute(ctx context.Context, tctx Context, name string, args // Avoid duplicate sends to the active conversation; reply with normal assistant text instead. if strings.TrimSpace(tctx.Channel) != "" && strings.TrimSpace(tctx.ChatID) != "" { if ch == strings.TrimSpace(tctx.Channel) && cid == strings.TrimSpace(tctx.ChatID) { - return "", errors.New("message to current session is not allowed; respond with assistant text instead") + return "", errors.New("message to current session is not allowed; use a different chat_id to send to another conversation/channel") + } + } + // Avoid duplicate sends to the active conversation; reply with normal assistant text instead. + if strings.TrimSpace(tctx.Channel) != "" && strings.TrimSpace(tctx.ChatID) != "" { + if ch == strings.TrimSpace(tctx.Channel) && cid == strings.TrimSpace(tctx.ChatID) { + return "", errors.New("message to current session is not allowed; use a different chat_id to send to another conversation/channel") } } return r.message(ctx, ch, cid, a.Content) + case "send_file": + var a struct { + FilePath string `json:"file_path"` + FileName string `json:"file_name"` + Channel string `json:"channel"` + ChatID string `json:"chat_id"` + Caption string `json:"caption"` + } + if err := json.Unmarshal(args, &a); err != nil { + return "", err + } + ch := strings.TrimSpace(a.Channel) + cid := strings.TrimSpace(a.ChatID) + if ch == "" || cid == "" || a.FilePath == "" { + return "", errors.New("send_file requires channel, chat_id, and file_path") + } + // For send_file, we allow sending to current session since files have different semantics than text messages + return r.sendFile(ctx, a.FilePath, a.FileName, ch, cid, a.Caption) case "spawn": var a struct { Task string `json:"task"` diff --git a/tools/tool_message.go b/tools/tool_message.go index cf03d66..127e331 100644 --- a/tools/tool_message.go +++ b/tools/tool_message.go @@ -26,3 +26,58 @@ func (r *Registry) message(ctx context.Context, channel, chatID, content string) } return fmt.Sprintf("Message sent to %s:%s", channel, chatID), nil } + +func (r *Registry) sendFile(ctx context.Context, filePath, fileName, channel, chatID, caption string) (string, error) { + filePath = strings.TrimSpace(filePath) + if filePath == "" { + return "", errors.New("file_path is empty") + } + if strings.TrimSpace(channel) == "" || strings.TrimSpace(chatID) == "" { + return "", errors.New("no target channel/chat_id") + } + if r.Outbound == nil { + return "", errors.New("file sending not configured") + } + + // Resolve file path relative to workspace + if !strings.HasPrefix(filePath, "/") { + filePath = strings.TrimPrefix(filePath, "./") + if r.WorkspaceDir != "" { + filePath = fmt.Sprintf("%s/%s", r.WorkspaceDir, filePath) + } + } + + // If fileName is not provided, use the base name of the file + if fileName == "" { + parts := strings.Split(filePath, "/") + fileName = parts[len(parts)-1] + } + + // Determine file kind based on extension or MIME type + kind := "file" + if strings.HasSuffix(strings.ToLower(fileName), ".jpg") || strings.HasSuffix(strings.ToLower(fileName), ".jpeg") || strings.HasSuffix(strings.ToLower(fileName), ".png") || strings.HasSuffix(strings.ToLower(fileName), ".gif") { + kind = "image" + } else if strings.HasSuffix(strings.ToLower(fileName), ".mp4") || strings.HasSuffix(strings.ToLower(fileName), ".avi") || strings.HasSuffix(strings.ToLower(fileName), ".mov") { + kind = "video" + } else if strings.HasSuffix(strings.ToLower(fileName), ".mp3") || strings.HasSuffix(strings.ToLower(fileName), ".wav") || strings.HasSuffix(strings.ToLower(fileName), ".ogg") { + kind = "audio" + } + + attachment := bus.Attachment{ + Name: fileName, + Kind: kind, + LocalPath: filePath, + } + + msg := bus.OutboundMessage{ + Channel: channel, + ChatID: chatID, + Content: strings.TrimSpace(caption), + Attachments: []bus.Attachment{attachment}, + } + + if err := r.Outbound(ctx, msg); err != nil { + return "", err + } + return fmt.Sprintf("File '%s' sent to %s:%s", fileName, channel, chatID), nil +} diff --git a/tools/tools_definitions_test.go b/tools/tools_definitions_test.go index 09c5445..f6f99a2 100644 --- a/tools/tools_definitions_test.go +++ b/tools/tools_definitions_test.go @@ -6,11 +6,17 @@ import ( "testing" "time" + "github.com/mosaxiv/clawlet/bus" "github.com/mosaxiv/clawlet/memory" ) type stubMemoryManager struct{} type stubSkillRegistry struct{} +type stubOutbound func(ctx context.Context, msg bus.OutboundMessage) error + +func stubOutboundFunc(ctx context.Context, msg bus.OutboundMessage) error { + return nil +} func (stubMemoryManager) Search(ctx context.Context, query string, opts memory.SearchOptions) ([]memory.SearchResult, error) { return []memory.SearchResult{}, nil @@ -119,3 +125,22 @@ func TestRegistryDefinitions_IncludesSkillRegistryToolsWhenEnabled(t *testing.T) } } } + +func TestRegistryDefinitions_IncludesMessageToolsWhenEnabled(t *testing.T) { + r := &Registry{ + WorkspaceDir: "/tmp", + RestrictToWorkspace: false, + ExecTimeout: 1 * time.Second, + Outbound: stubOutboundFunc, + } + defs := r.Definitions() + has := map[string]bool{} + for _, d := range defs { + has[d.Function.Name] = true + } + for _, n := range []string{"message", "send_file"} { + if !has[n] { + t.Fatalf("expected message tool definition: %s", n) + } + } +}