From ead9c0d3e58d8ef9c46e9f8959df4cdeba5e06f7 Mon Sep 17 00:00:00 2001 From: Worm Date: Wed, 18 Feb 2026 13:32:28 +0800 Subject: [PATCH] add chat state --- README.zh.md | 7 +++ pkg/agent/loop.go | 8 ++-- pkg/channels/manager.go | 19 +++++++++ pkg/channels/xmpp.go | 94 ++++++++++++++++++++++++++++++++++++++++- 4 files changed, 123 insertions(+), 5 deletions(-) diff --git a/README.zh.md b/README.zh.md index e0028e7c9..6aeca2794 100644 --- a/README.zh.md +++ b/README.zh.md @@ -500,6 +500,13 @@ PicoClaw 将数据存储在您配置的工作区中（默认：`~/.picoclaw/work - 对安全敏感的 XMPP 渠道设置较短 TTL（如 `1h`），降低明文历史泄露风险 - 对相对不敏感或需要长期上下文的渠道保留 `"false"`，完全不自动清理 +对于 XMPP 渠道，Agent 会默认： + +- 在处理用户消息时发送「正在输入 / active」状态（XEP‑0085），方便前端展示输入指示； +- 在收到带 `` 的消息时自动发送 `` 回执（XEP‑0184）。 + +这两个行为都是**默认开启且向后兼容**的：旧版 `config.json` 不需要增加任何字段即可使用，新版客户端若不支持相应 XEP 也会直接忽略这些附加元素。 + ### 心跳 / 周期性任务 (Heartbeat) PicoClaw 可以自动执行周期性任务。在工作区创建 `HEARTBEAT.md` 文件： diff --git a/pkg/agent/loop.go b/pkg/agent/loop.go index f3ccba725..bcc7dcc61 100644 --- a/pkg/agent/loop.go +++ b/pkg/agent/loop.go @@ -275,17 +275,19 @@ func (al *AgentLoop) processMessage(ctx context.Context, msg bus.InboundMessage) "session_key": msg.SessionKey, }) - // Route system messages to processSystemMessage if msg.Channel == "system" { return al.processSystemMessage(ctx, msg) } - // Check for commands if response, handled := al.handleCommand(ctx, msg); handled { return response, nil } - // Process as user message + if al.channelManager != nil && msg.Channel != "" && msg.ChatID != "" && !constants.IsInternalChannel(msg.Channel) { + al.channelManager.NotifyTyping(ctx, msg.Channel, msg.ChatID, true) + defer al.channelManager.NotifyTyping(ctx, msg.Channel, msg.ChatID, false) + } + return al.runAgentLoop(ctx, processOptions{ SessionKey: msg.SessionKey, Channel: msg.Channel, diff --git a/pkg/channels/manager.go b/pkg/channels/manager.go index b47e957d9..c167b6131 100644 --- a/pkg/channels/manager.go +++ b/pkg/channels/manager.go @@ -17,6 +17,11 @@ import ( "github.com/sipeed/picoclaw/pkg/logger" ) +type TypingChannel interface { + Channel + SendTyping(ctx context.Context, chatID string, composing bool) error +} + type Manager struct { channels map[string]Channel bus *bus.MessageBus @@ -339,6 +344,20 @@ func (m *Manager) UnregisterChannel(name string) { delete(m.channels, name) } +func (m *Manager) NotifyTyping(ctx context.Context, channelName, chatID string, composing bool) { + m.mu.RLock() + channel, exists := m.channels[channelName] + m.mu.RUnlock() + if !exists { + return + } + typingChannel, ok := channel.(TypingChannel) + if !ok { + return + } + _ = typingChannel.SendTyping(ctx, chatID, composing) +} + func (m *Manager) SendToChannel(ctx context.Context, channelName, chatID, content string) error { m.mu.RLock() channel, exists := m.channels[channelName] diff --git a/pkg/channels/xmpp.go b/pkg/channels/xmpp.go index da742ea8e..8ccdb5623 100644 --- a/pkg/channels/xmpp.go +++ b/pkg/channels/xmpp.go @@ -47,6 +47,9 @@ type XMPPChannel struct { lastTime time.Time } +const chatStatesNS = "http://jabber.org/protocol/chatstates" +const receiptsNS = "urn:xmpp:receipts" + func NewXMPPChannel(cfg config.XMPPConfig, messageBus *bus.MessageBus) (*XMPPChannel, error) { if cfg.JID == "" { return nil, fmt.Errorf("xmpp jid is required") @@ -269,12 +272,24 @@ func (c *XMPPChannel) sendChatMessage(to jid.JID, body string, desc string, medi payload = xmlstream.MultiReader(children...) } + stateElem := xmlstream.Wrap( + nil, + xml.StartElement{ + Name: xml.Name{Local: "active"}, + Attr: []xml.Attr{ + {Name: xml.Name{Local: "xmlns"}, Value: chatStatesNS}, + }, + }, + ) + + fullPayload := xmlstream.MultiReader(payload, stateElem) + st := stanza.Message{ Type: stanza.ChatMessage, To: to, } - return c.sendStanza(st.Wrap(payload)) + return c.sendStanza(st.Wrap(fullPayload)) } func (c *XMPPChannel) sendStanza(r xml.TokenReader) error { @@ -302,7 +317,10 @@ func (c *XMPPChannel) sendStanza(r xml.TokenReader) error { func (c *XMPPChannel) handleIncomingMessage(msg stanza.Message, t xmlstream.TokenReadEncoder) error { var payload struct { - Body string `xml:"body"` + Body string `xml:"body"` + Request *struct { + XMLName xml.Name `xml:"urn:xmpp:receipts request"` + } `xml:"urn:xmpp:receipts request"` } d := xml.NewTokenDecoder(t) @@ -312,9 +330,16 @@ func (c *XMPPChannel) handleIncomingMessage(msg stanza.Message, t xmlstream.Toke content := strings.TrimSpace(payload.Body) if content == "" { + if payload.Request != nil && msg.ID != "" { + go c.sendDeliveryReceipt(msg) + } return nil } + if payload.Request != nil && msg.ID != "" { + go c.sendDeliveryReceipt(msg) + } + fromBare := msg.From.Bare().String() chatID := msg.From.String() @@ -342,6 +367,71 @@ func (c *XMPPChannel) handleIncomingMessage(msg stanza.Message, t xmlstream.Toke return nil } +func (c *XMPPChannel) sendChatState(to jid.JID, state string) error { + if !c.IsRunning() || c.session == nil { + return nil + } + + elem := xmlstream.Wrap( + nil, + xml.StartElement{ + Name: xml.Name{Local: state}, + Attr: []xml.Attr{ + {Name: xml.Name{Local: "xmlns"}, Value: chatStatesNS}, + }, + }, + ) + + msg := stanza.Message{ + Type: stanza.ChatMessage, + To: to, + } + + return c.sendStanza(msg.Wrap(elem)) +} + +func (c *XMPPChannel) SendTyping(ctx context.Context, chatID string, composing bool) error { + if !c.IsRunning() || c.session == nil { + return nil + } + + to, err := jid.Parse(chatID) + if err != nil { + return err + } + + state := "active" + if composing { + state = "composing" + } + + return c.sendChatState(to, state) +} + +func (c *XMPPChannel) sendDeliveryReceipt(orig stanza.Message) error { + if !c.IsRunning() || c.session == nil || orig.ID == "" { + return nil + } + + payload := xmlstream.Wrap( + nil, + xml.StartElement{ + Name: xml.Name{Local: "received"}, + Attr: []xml.Attr{ + {Name: xml.Name{Local: "xmlns"}, Value: receiptsNS}, + {Name: xml.Name{Local: "id"}, Value: orig.ID}, + }, + }, + ) + + reply := stanza.Message{ + Type: orig.Type, + To: orig.From, + } + + return c.sendStanza(reply.Wrap(payload)) +} + func (c *XMPPChannel) discoverUploadService(ctx context.Context) (jid.JID, error) { c.uploadMu.Lock() defer c.uploadMu.Unlock()