feat:add imsg channel

config/config.example.json

@@ -110,6 +110,11 @@
       },
       "reasoning_channel_id": ""
     },
+    "imsg": {
+      "enabled": false,
+      "allow_from": [],
+      "iMessageCLIPath": "imsg"
+    },
     "qq": {
       "enabled": false,
       "app_id": "YOUR_QQ_APP_ID",

pkg/channels/imsg/imsg.go

@@ -0,0 +1,824 @@
+package imsg
+
+import (
+	"bufio"
+	"context"
+	"crypto/sha1"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os/exec"
+	"reflect"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"github.com/sipeed/picoclaw/pkg/bus"
+	"github.com/sipeed/picoclaw/pkg/channels"
+	"github.com/sipeed/picoclaw/pkg/config"
+	"github.com/sipeed/picoclaw/pkg/identity"
+	"github.com/sipeed/picoclaw/pkg/logger"
+)
+
+type IMsgChannel struct {
+	*channels.BaseChannel
+	config    config.IMsgConfig
+	runCtx    context.Context
+	runCancel context.CancelFunc
+	rpcCmd    *exec.Cmd
+	rpcStdin  io.WriteCloser
+	wg        sync.WaitGroup
+	mu        sync.Mutex
+	fatalMu   sync.Mutex
+	fatalLine string
+	lastLine  string
+	recvMode  string
+	seenMu    sync.Mutex
+	seenIDs   map[string]struct{}
+	seenEvent map[string]time.Time
+	reqSeq    atomic.Uint64
+	writeMu   sync.Mutex
+	pendingMu sync.Mutex
+	pending   map[string]chan rpcResponse
+}
+
+func NewIMsgChannel(cfg config.IMsgConfig, messageBus *bus.MessageBus) (*IMsgChannel, error) {
+	if strings.TrimSpace(cfg.IMessageCLIPath) == "" {
+		cfg.IMessageCLIPath = "imsg"
+	}
+
+	base := channels.NewBaseChannel("imsg", cfg, messageBus, cfg.AllowFrom)
+	return &IMsgChannel{
+		BaseChannel: base,
+		config:      cfg,
+		seenIDs:     make(map[string]struct{}),
+		seenEvent:   make(map[string]time.Time),
+		pending:     make(map[string]chan rpcResponse),
+	}, nil
+}
+
+type rpcResponse struct {
+	Result any
+	Error  *rpcError
+}
+
+type rpcError struct {
+	Code    int    `json:"code"`
+	Message string `json:"message"`
+}
+
+func (c *IMsgChannel) Start(ctx context.Context) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if c.IsRunning() {
+		return nil
+	}
+	c.fatalMu.Lock()
+	c.fatalLine = ""
+	c.lastLine = ""
+	c.fatalMu.Unlock()
+	c.runCtx, c.runCancel = context.WithCancel(ctx)
+	if err := c.startRPCLocked(); err != nil {
+		c.runCancel()
+		c.runCtx = nil
+		c.runCancel = nil
+		return err
+	}
+	c.SetRunning(true)
+	logger.InfoC("imsg", "iMessage channel started")
+	return nil
+}
+
+func (c *IMsgChannel) Stop(ctx context.Context) error {
+	c.mu.Lock()
+	if !c.IsRunning() && c.rpcCmd == nil {
+		c.mu.Unlock()
+		return nil
+	}
+	c.SetRunning(false)
+	if c.runCancel != nil {
+		c.runCancel()
+	}
+	cmd := c.rpcCmd
+	stdin := c.rpcStdin
+	c.rpcCmd = nil
+	c.rpcStdin = nil
+	c.mu.Unlock()
+
+	if stdin != nil {
+		_ = stdin.Close()
+	}
+	if cmd != nil && cmd.Process != nil {
+		_ = cmd.Process.Kill()
+	}
+	c.wg.Wait()
+
+	c.mu.Lock()
+	c.runCtx = nil
+	c.runCancel = nil
+	c.mu.Unlock()
+
+	logger.InfoC("imsg", "iMessage channel stopped")
+	return nil
+}
+
+func (c *IMsgChannel) Send(ctx context.Context, msg bus.OutboundMessage) ([]string, error) {
+	if !c.IsRunning() {
+		return nil, channels.ErrNotRunning
+	}
+	paramsList := []map[string]any{
+		{"chat_id": msg.ChatID, "text": msg.Content},
+		{"chatId": msg.ChatID, "text": msg.Content},
+		{"chat_id": msg.ChatID, "content": msg.Content},
+	}
+	methods := []string{"send", "message.send"}
+	var lastErr error
+	for _, method := range methods {
+		for _, params := range paramsList {
+			logger.InfoCF("imsg", "imsg rpc send request", map[string]any{
+				"method": method,
+				"params": params,
+			})
+			if _, err := c.rpcCall(ctx, method, params, 8*time.Second); err == nil {
+				return nil, nil
+			} else {
+				lastErr = err
+			}
+		}
+	}
+	return nil, fmt.Errorf("imsg rpc send failed: %w", lastErr)
+}
+
+// SendMedia implements channels.MediaSender via imsg RPC.
+func (c *IMsgChannel) SendMedia(ctx context.Context, msg bus.OutboundMediaMessage) ([]string, error) {
+	if !c.IsRunning() {
+		return nil, channels.ErrNotRunning
+	}
+	store := c.GetMediaStore()
+	if store == nil {
+		return nil, fmt.Errorf("no media store available: %w", channels.ErrSendFailed)
+	}
+
+	for _, part := range msg.Parts {
+		localPath, err := store.Resolve(part.Ref)
+		if err != nil {
+			return nil, fmt.Errorf("resolve media ref %s: %w", part.Ref, err)
+		}
+		paramsList := []map[string]any{
+			{"chat_id": msg.ChatID, "text": part.Caption, "file": localPath},
+			{"chatId": msg.ChatID, "text": part.Caption, "file": localPath},
+			{"chat_id": msg.ChatID, "content": part.Caption, "file": localPath},
+		}
+		methods := []string{"send", "message.send"}
+		var lastErr error
+		ok := false
+		for _, method := range methods {
+			for _, params := range paramsList {
+				logger.InfoCF("imsg", "imsg rpc send media request", map[string]any{
+					"method": method,
+					"params": params,
+				})
+				if _, err := c.rpcCall(ctx, method, params, 12*time.Second); err == nil {
+					ok = true
+					break
+				} else {
+					lastErr = err
+				}
+			}
+			if ok {
+				break
+			}
+		}
+		if !ok {
+			return nil, fmt.Errorf("imsg rpc send media failed: %w", lastErr)
+		}
+	}
+	return nil, nil
+}
+
+func (c *IMsgChannel) startRPCLocked() error {
+	cliPath := strings.TrimSpace(c.config.IMessageCLIPath)
+	if cliPath == "" {
+		cliPath = "imsg"
+	}
+
+	commands := []struct {
+		mode string
+		args []string
+	}{
+		{mode: "rpc", args: []string{"rpc", "--json"}},
+		{mode: "rpc", args: []string{"rpc"}},
+	}
+	var lastErr error
+	for _, candidate := range commands {
+		if err := c.startReceiveProcessLocked(cliPath, candidate.mode, candidate.args); err != nil {
+			lastErr = err
+			continue
+		}
+		if err := c.subscribeRPC(c.runCtx); err != nil {
+			c.cleanupReceiveProcessLocked()
+			lastErr = err
+			continue
+		}
+		return nil
+	}
+	if lastErr != nil {
+		return lastErr
+	}
+	return fmt.Errorf("imsg startup failed: no receive command available")
+}
+
+func (c *IMsgChannel) cleanupReceiveProcessLocked() {
+	stdin := c.rpcStdin
+	cmd := c.rpcCmd
+	c.rpcStdin = nil
+	c.rpcCmd = nil
+	if stdin != nil {
+		_ = stdin.Close()
+	}
+	if cmd != nil && cmd.Process != nil {
+		_ = cmd.Process.Kill()
+	}
+	c.wg.Wait()
+}
+
+func (c *IMsgChannel) startReceiveProcessLocked(cliPath, mode string, args []string) error {
+	cmd := exec.CommandContext(c.runCtx, cliPath, args...)
+	logger.InfoCF("imsg", "imsg receive command", map[string]any{
+		"mode": mode,
+		"cmd":  strings.Join(cmd.Args, " "),
+	})
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return fmt.Errorf("imsg %s stdout pipe: %w", mode, err)
+	}
+	stderr, err := cmd.StderrPipe()
+	if err != nil {
+		return fmt.Errorf("imsg %s stderr pipe: %w", mode, err)
+	}
+	stdin, err := cmd.StdinPipe()
+	if err != nil {
+		return fmt.Errorf("imsg %s stdin pipe: %w", mode, err)
+	}
+	if err := cmd.Start(); err != nil {
+		return fmt.Errorf("start imsg %s: %w", mode, err)
+	}
+	c.rpcCmd = cmd
+	c.rpcStdin = stdin
+	c.recvMode = mode
+
+	exited := make(chan error, 1)
+	c.wg.Add(3)
+	go c.consumeRPCStdout(stdout)
+	go c.consumeRPCStderr(stderr)
+	go c.waitRPCExit(cmd, exited)
+
+	// Startup probe: if RPC exits immediately, fail Start with a clear reason.
+	select {
+	case err := <-exited:
+		c.rpcCmd = nil
+		c.rpcStdin = nil
+		c.wg.Wait()
+		if err == nil {
+			if fatal := c.getFatalLine(); fatal != "" {
+				return fmt.Errorf(
+					"imsg %s exited during startup (cmd=%q, fatal=%s)",
+					mode,
+					strings.Join(cmd.Args, " "),
+					fatal,
+				)
+			}
+			if last := c.getLastLine(); last != "" {
+				return fmt.Errorf(
+					"imsg %s exited during startup (cmd=%q, last output: %s)",
+					mode,
+					strings.Join(cmd.Args, " "),
+					last,
+				)
+			}
+			return fmt.Errorf(
+				"imsg %s exited during startup (cmd=%q, no output captured)",
+				mode,
+				strings.Join(cmd.Args, " "),
+			)
+		}
+		if fatal := c.getFatalLine(); fatal != "" {
+			return fmt.Errorf(
+				"imsg %s startup failed: %v (cmd=%q, fatal=%s)",
+				mode,
+				err,
+				strings.Join(cmd.Args, " "),
+				fatal,
+			)
+		}
+		if last := c.getLastLine(); last != "" {
+			return fmt.Errorf(
+				"imsg %s startup failed: %v (cmd=%q, last output: %s)",
+				mode,
+				err,
+				strings.Join(cmd.Args, " "),
+				last,
+			)
+		}
+		return fmt.Errorf(
+			"imsg %s startup failed: %w (cmd=%q, no output captured)",
+			mode,
+			err,
+			strings.Join(cmd.Args, " "),
+		)
+	case <-time.After(1200 * time.Millisecond):
+	}
+	logger.InfoCF("imsg", "imsg receive process started", map[string]any{
+		"mode": mode,
+		"cmd":  strings.Join(cmd.Args, " "),
+	})
+	// NOTE: History polling fallback is intentionally disabled.
+	// Use pure JSON-RPC watch.subscribe for inbound messages.
+	return nil
+}
+
+func (c *IMsgChannel) consumeRPCStdout(r io.Reader) {
+	defer c.wg.Done()
+	scanner := bufio.NewScanner(r)
+	scanner.Buffer(make([]byte, 0, 64*1024), 2*1024*1024)
+	for scanner.Scan() {
+		c.handleRPCLine(scanner.Text())
+	}
+	if err := scanner.Err(); err != nil && c.IsRunning() {
+		logger.WarnCF("imsg", "imsg rpc stdout scanner error", map[string]any{"error": err.Error()})
+	}
+}
+
+func (c *IMsgChannel) consumeRPCStderr(r io.Reader) {
+	defer c.wg.Done()
+	scanner := bufio.NewScanner(r)
+	scanner.Buffer(make([]byte, 0, 8*1024), 512*1024)
+	for scanner.Scan() {
+		line := strings.TrimSpace(scanner.Text())
+		if line == "" {
+			continue
+		}
+		c.setLastLine(line)
+		if isFatalRPCLine(line) {
+			c.setFatalLine(line)
+		}
+		logger.DebugCF("imsg", "imsg rpc stderr", map[string]any{"line": line})
+	}
+}
+
+func (c *IMsgChannel) waitRPCExit(cmd *exec.Cmd, exited chan<- error) {
+	defer c.wg.Done()
+	err := cmd.Wait()
+	select {
+	case exited <- err:
+	default:
+	}
+	if err != nil && c.IsRunning() {
+		fields := map[string]any{"error": err.Error()}
+		if fatal := c.getFatalLine(); fatal != "" {
+			fields["hint"] = fatal
+		} else if last := c.getLastLine(); last != "" {
+			fields["last_output"] = last
+		}
+		logger.ErrorCF("imsg", "imsg rpc exited unexpectedly", fields)
+	}
+}
+
+func (c *IMsgChannel) handleRPCLine(line string) {
+	line = strings.TrimSpace(line)
+	if line == "" {
+		return
+	}
+	c.setLastLine(line)
+	var raw map[string]any
+	if err := json.Unmarshal([]byte(line), &raw); err != nil {
+		if isFatalRPCLine(line) {
+			c.setFatalLine(line)
+			logger.ErrorCF("imsg", "imsg rpc fatal output", map[string]any{"line": line})
+			return
+		}
+		logger.DebugCF("imsg", "skip non-json rpc line", map[string]any{"line": line})
+		return
+	}
+	// Handle JSON-RPC responses first.
+	if id, ok := raw["id"]; ok && (raw["result"] != nil || raw["error"] != nil) {
+		c.resolvePending(id, raw)
+		return
+	}
+
+	method := strings.ToLower(anyToString(raw["method"]))
+	if method != "" {
+		// imsg legacy notification: method = "message"
+		if strings.Contains(method, "message") || strings.Contains(method, "watch") {
+			params := firstMap(raw["params"], raw["result"], raw["payload"], raw["message"], raw)
+			c.handlePayload(params, line)
+			return
+		}
+		// Non-message RPC notifications (e.g. ping/ack) should not enter agent loop.
+		logger.DebugCF("imsg", "skip non-message rpc notification", map[string]any{"method": method})
+		return
+	}
+
+	// Ignore JSON lines without method to avoid parsing RPC response-like payloads
+	// as inbound chat messages, which can create self-reply loops.
+	logger.DebugCF("imsg", "skip rpc json without method", map[string]any{"line": line})
+}
+
+func (c *IMsgChannel) handlePayload(raw map[string]any, rawLine string) {
+	// Ignore messages sent by ourselves to avoid self-trigger loops.
+	if deepBool(raw, "is_from_me", "from_me", "is_me", "isFromMe", "fromMe", "outgoing") {
+		return
+	}
+	content := deepCoalesceString(raw, "content", "text", "message", "body")
+	if strings.TrimSpace(content) == "" {
+		return
+	}
+
+	kind := strings.ToLower(deepCoalesceString(raw, "type", "event", "method", "name"))
+	if strings.Contains(kind, "outgoing") || strings.Contains(kind, "sent") {
+		return
+	}
+
+	senderID := deepCoalesceString(raw, "from", "sender", "sender_id", "handle", "from_id", "participant")
+	chatID := deepCoalesceString(
+		raw,
+		"chat_id",
+		"chat",
+		"conversation",
+		"thread",
+		"peer",
+		"to",
+		"chat_identifier",
+		"conversation_id",
+	)
+	if senderID == "" && chatID == "" {
+		logger.DebugCF("imsg", "imsg rpc json parsed but missing sender/chat", map[string]any{
+			"kind": kind,
+			"line": rawLine,
+		})
+		return
+	}
+	if senderID == "" {
+		senderID = chatID
+	}
+	if chatID == "" {
+		chatID = senderID
+	}
+	messageID := deepCoalesceString(raw, "id", "message_id", "guid", "rowid")
+	if messageID != "" && !c.markSeen(messageID) {
+		return
+	}
+	if messageID == "" {
+		fingerprint := eventFingerprint(chatID, senderID, content, deepCoalesceString(raw, "timestamp", "date", "time"))
+		if !c.markSeenEvent(fingerprint) {
+			return
+		}
+	}
+
+	peerKind := "direct"
+	if deepBool(raw, "is_group", "group", "isGroup") {
+		peerKind = "group"
+	}
+	peer := bus.Peer{Kind: peerKind, ID: chatID}
+	sender := bus.SenderInfo{
+		Platform:    "imsg",
+		PlatformID:  senderID,
+		CanonicalID: identity.BuildCanonicalID("imsg", senderID),
+	}
+	if !c.isAllowedIMsg(sender, senderID, chatID) {
+		logger.DebugCF("imsg", "imsg message blocked by allow_from", map[string]any{
+			"sender_id": senderID,
+			"chat_id":   chatID,
+		})
+		c.sendNotAllowedNotice(chatID)
+		return
+	}
+	metadata := map[string]string{"platform": "imsg"}
+	if kind != "" {
+		metadata["rpc_event"] = kind
+	}
+	metadata["recv_mode"] = c.recvMode
+	if rawLine != "" {
+		metadata["raw_line"] = rawLine
+	}
+	logger.InfoCF("imsg", "imsg inbound message received", map[string]any{
+		"mode":      c.recvMode,
+		"event":     kind,
+		"sender_id": senderID,
+		"chat_id":   chatID,
+	})
+	c.HandleMessage(c.runCtx, peer, messageID, senderID, chatID, content, nil, metadata, sender)
+}
+
+func (c *IMsgChannel) isAllowedIMsg(sender bus.SenderInfo, senderID, chatID string) bool {
+	// 1) Canonical sender matching (preferred).
+	if c.IsAllowedSender(sender) {
+		return true
+	}
+	// 2) Legacy sender matching.
+	if senderID != "" && c.IsAllowed(senderID) {
+		return true
+	}
+	return false
+}
+
+func (c *IMsgChannel) sendNotAllowedNotice(chatID string) {
+	if strings.TrimSpace(chatID) == "" || !c.IsRunning() {
+		return
+	}
+	_, err := c.Send(c.runCtx, bus.OutboundMessage{
+		Channel: "imsg",
+		ChatID:  chatID,
+		Content: "not allowed",
+	})
+	if err != nil {
+		logger.DebugCF("imsg", "failed to send not-allowed notice", map[string]any{
+			"chat_id": chatID,
+			"error":   err.Error(),
+		})
+	}
+}
+
+func deepCoalesceString(v any, keys ...string) string {
+	for _, k := range keys {
+		if s := deepLookupString(v, k); s != "" {
+			return s
+		}
+	}
+	return ""
+}
+
+func deepLookupString(v any, key string) string {
+	switch vv := v.(type) {
+	case map[string]any:
+		if val, ok := vv[key]; ok {
+			if s := anyToString(val); s != "" {
+				return s
+			}
+		}
+		for _, child := range vv {
+			if s := deepLookupString(child, key); s != "" {
+				return s
+			}
+		}
+	case []any:
+		for _, child := range vv {
+			if s := deepLookupString(child, key); s != "" {
+				return s
+			}
+		}
+	}
+	return ""
+}
+
+func anyToString(v any) string {
+	switch vv := v.(type) {
+	case string:
+		if strings.TrimSpace(vv) != "" {
+			return vv
+		}
+	case json.Number:
+		return vv.String()
+	case float64:
+		return fmt.Sprintf("%.0f", vv)
+	case int, int32, int64, uint, uint32, uint64:
+		return fmt.Sprintf("%v", vv)
+	}
+	return ""
+}
+
+func deepBool(v any, keys ...string) bool {
+	for _, k := range keys {
+		if b, ok := deepLookupBool(v, k); ok {
+			return b
+		}
+	}
+	return false
+}
+
+func deepLookupBool(v any, key string) (bool, bool) {
+	switch vv := v.(type) {
+	case map[string]any:
+		if val, ok := vv[key]; ok {
+			if b, okb := val.(bool); okb {
+				return b, true
+			}
+			// Handle 0/1 style flags.
+			rv := reflect.ValueOf(val)
+			switch rv.Kind() {
+			case reflect.Int, reflect.Int32, reflect.Int64:
+				return rv.Int() != 0, true
+			case reflect.Uint, reflect.Uint32, reflect.Uint64:
+				return rv.Uint() != 0, true
+			case reflect.Float32, reflect.Float64:
+				return rv.Float() != 0, true
+			}
+		}
+		for _, child := range vv {
+			if b, ok := deepLookupBool(child, key); ok {
+				return b, true
+			}
+		}
+	case []any:
+		for _, child := range vv {
+			if b, ok := deepLookupBool(child, key); ok {
+				return b, true
+			}
+		}
+	}
+	return false, false
+}
+
+func (c *IMsgChannel) subscribeRPC(ctx context.Context) error {
+	methods := []string{"watch.subscribe", "subscribe"}
+	var lastErr error
+	for _, m := range methods {
+		logger.InfoCF("imsg", "imsg rpc subscribe request", map[string]any{"method": m})
+		if _, err := c.rpcCall(ctx, m, map[string]any{}, 6*time.Second); err == nil {
+			return nil
+		} else {
+			lastErr = err
+		}
+	}
+	return fmt.Errorf("rpc subscribe failed: %w", lastErr)
+}
+
+func (c *IMsgChannel) rpcCall(ctx context.Context, method string, params any, timeout time.Duration) (any, error) {
+	id := fmt.Sprintf("%d", c.reqSeq.Add(1))
+	req := map[string]any{
+		"jsonrpc": "2.0",
+		"id":      id,
+		"method":  method,
+		"params":  params,
+	}
+	data, err := json.Marshal(req)
+	if err != nil {
+		return nil, err
+	}
+
+	respCh := make(chan rpcResponse, 1)
+	c.pendingMu.Lock()
+	c.pending[id] = respCh
+	c.pendingMu.Unlock()
+	defer func() {
+		c.pendingMu.Lock()
+		delete(c.pending, id)
+		c.pendingMu.Unlock()
+	}()
+
+	c.writeMu.Lock()
+	stdin := c.rpcStdin
+	if stdin == nil {
+		c.writeMu.Unlock()
+		return nil, fmt.Errorf("rpc stdin unavailable")
+	}
+	_, wErr := io.WriteString(stdin, string(data)+"\n")
+	c.writeMu.Unlock()
+	if wErr != nil {
+		return nil, fmt.Errorf("rpc write failed: %w", wErr)
+	}
+
+	waitCtx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+	select {
+	case <-waitCtx.Done():
+		return nil, waitCtx.Err()
+	case resp := <-respCh:
+		if resp.Error != nil {
+			return nil, fmt.Errorf("rpc error %d: %s", resp.Error.Code, resp.Error.Message)
+		}
+		return resp.Result, nil
+	}
+}
+
+func (c *IMsgChannel) resolvePending(id any, raw map[string]any) {
+	key := anyToString(id)
+	if key == "" {
+		return
+	}
+	c.pendingMu.Lock()
+	ch, ok := c.pending[key]
+	c.pendingMu.Unlock()
+	if !ok {
+		return
+	}
+	resp := rpcResponse{Result: raw["result"]}
+	if errObj, ok := raw["error"].(map[string]any); ok {
+		resp.Error = &rpcError{
+			Code:    int(toFloat(errObj["code"])),
+			Message: anyToString(errObj["message"]),
+		}
+	}
+	select {
+	case ch <- resp:
+	default:
+	}
+}
+
+func firstMap(candidates ...any) map[string]any {
+	for _, v := range candidates {
+		switch vv := v.(type) {
+		case map[string]any:
+			if msg, ok := vv["message"].(map[string]any); ok {
+				return msg
+			}
+			if data, ok := vv["data"].(map[string]any); ok {
+				return data
+			}
+			if payload, ok := vv["payload"].(map[string]any); ok {
+				return payload
+			}
+			return vv
+		case []any:
+			for _, item := range vv {
+				if mm, ok := item.(map[string]any); ok {
+					return mm
+				}
+			}
+		}
+	}
+	return map[string]any{}
+}
+
+func toFloat(v any) float64 {
+	switch vv := v.(type) {
+	case float64:
+		return vv
+	case int:
+		return float64(vv)
+	case int64:
+		return float64(vv)
+	case json.Number:
+		f, _ := vv.Float64()
+		return f
+	}
+	return 0
+}
+
+func isFatalRPCLine(line string) bool {
+	l := strings.ToLower(line)
+	return strings.Contains(l, "permissiondenied") ||
+		strings.Contains(l, "authorization denied") ||
+		(strings.Contains(l, "chat.db") && strings.Contains(l, "denied"))
+}
+
+func (c *IMsgChannel) setFatalLine(line string) {
+	c.fatalMu.Lock()
+	defer c.fatalMu.Unlock()
+	if c.fatalLine == "" {
+		c.fatalLine = line
+	}
+}
+
+func (c *IMsgChannel) getFatalLine() string {
+	c.fatalMu.Lock()
+	defer c.fatalMu.Unlock()
+	return c.fatalLine
+}
+
+func (c *IMsgChannel) setLastLine(line string) {
+	c.fatalMu.Lock()
+	defer c.fatalMu.Unlock()
+	c.lastLine = line
+}
+
+func (c *IMsgChannel) getLastLine() string {
+	c.fatalMu.Lock()
+	defer c.fatalMu.Unlock()
+	return c.lastLine
+}
+
+func (c *IMsgChannel) markSeen(id string) bool {
+	c.seenMu.Lock()
+	defer c.seenMu.Unlock()
+	if _, ok := c.seenIDs[id]; ok {
+		return false
+	}
+	c.seenIDs[id] = struct{}{}
+	return true
+}
+
+func (c *IMsgChannel) markSeenEvent(fp string) bool {
+	c.seenMu.Lock()
+	defer c.seenMu.Unlock()
+	now := time.Now()
+	const ttl = 2 * time.Minute
+	if t, ok := c.seenEvent[fp]; ok && now.Sub(t) <= ttl {
+		return false
+	}
+	c.seenEvent[fp] = now
+	// Opportunistic cleanup to keep map bounded.
+	for k, t := range c.seenEvent {
+		if now.Sub(t) > ttl {
+			delete(c.seenEvent, k)
+		}
+	}
+	return true
+}
+
+func eventFingerprint(chatID, senderID, content, ts string) string {
+	sum := sha1.Sum([]byte(chatID + "|" + senderID + "|" + content + "|" + ts))
+	return fmt.Sprintf("%x", sum[:])
+}

pkg/channels/imsg/init.go

@@ -0,0 +1,16 @@
+package imsg
+
+import (
+	"github.com/sipeed/picoclaw/pkg/bus"
+	"github.com/sipeed/picoclaw/pkg/channels"
+	"github.com/sipeed/picoclaw/pkg/config"
+)
+
+func init() {
+	channels.RegisterFactory("imsg", func(cfg *config.Config, b *bus.MessageBus) (channels.Channel, error) {
+		if !cfg.Channels.IMsg.Enabled {
+			return nil, nil
+		}
+		return NewIMsgChannel(cfg.Channels.IMsg, b)
+	})
+}

pkg/channels/manager.go

@@ -375,6 +375,10 @@ func (m *Manager) initChannels(channels *config.ChannelsConfig) error {
 		m.initChannel("discord", "Discord")
 	}
 
+	if channels.IMsg.Enabled {
+		m.initChannel("imsg", "iMessage")
+	}
+
 	if channels.MaixCam.Enabled {
 		m.initChannel("maixcam", "MaixCam")
 	}

pkg/config/config.go

@@ -284,6 +284,7 @@ type ChannelsConfig struct {
 	Telegram   TelegramConfig   `json:"telegram"    yaml:"telegram,omitempty"`
 	Feishu     FeishuConfig     `json:"feishu"      yaml:"feishu,omitempty"`
 	Discord    DiscordConfig    `json:"discord"     yaml:"discord,omitempty"`
+	IMsg       IMsgConfig       `json:"imsg"        yaml:"imsg,omitempty"`
 	MaixCam    MaixCamConfig    `json:"maixcam"     yaml:"-"`
 	QQ         QQConfig         `json:"qq"          yaml:"qq,omitempty"`
 	DingTalk   DingTalkConfig   `json:"dingtalk"    yaml:"dingtalk,omitempty"`
@@ -386,6 +387,12 @@ type DiscordConfig struct {
 	ReasoningChannelID string              `json:"reasoning_channel_id"    yaml:"-"               env:"PICOCLAW_CHANNELS_DISCORD_REASONING_CHANNEL_ID"`
 }
 
+type IMsgConfig struct {
+	Enabled         bool                `json:"enabled"         yaml:"-" env:"PICOCLAW_CHANNELS_IMSG_ENABLED"`
+	AllowFrom       FlexibleStringSlice `json:"allow_from"      yaml:"-" env:"PICOCLAW_CHANNELS_IMSG_ALLOW_FROM"`
+	IMessageCLIPath string              `json:"iMessageCLIPath" yaml:"-" env:"PICOCLAW_CHANNELS_IMSG_IMESSAGECLIPATH"`
+}
+
 type MaixCamConfig struct {
 	Enabled            bool                `json:"enabled"              env:"PICOCLAW_CHANNELS_MAIXCAM_ENABLED"`
 	Host               string              `json:"host"                 env:"PICOCLAW_CHANNELS_MAIXCAM_HOST"`

pkg/config/defaults.go

@@ -68,6 +68,11 @@ func DefaultConfig() *Config {
 				AllowFrom:   FlexibleStringSlice{},
 				MentionOnly: false,
 			},
+			IMsg: IMsgConfig{
+				Enabled:         false,
+				AllowFrom:       FlexibleStringSlice{},
+				IMessageCLIPath: "imsg",
+			},
 			MaixCam: MaixCamConfig{
 				Enabled:   false,
 				Host:      "0.0.0.0",

pkg/gateway/gateway.go

@@ -21,6 +21,7 @@ import (
 	_ "github.com/sipeed/picoclaw/pkg/channels/dingtalk"
 	_ "github.com/sipeed/picoclaw/pkg/channels/discord"
 	_ "github.com/sipeed/picoclaw/pkg/channels/feishu"
+	_ "github.com/sipeed/picoclaw/pkg/channels/imsg"
 	_ "github.com/sipeed/picoclaw/pkg/channels/irc"
 	_ "github.com/sipeed/picoclaw/pkg/channels/line"
 	_ "github.com/sipeed/picoclaw/pkg/channels/maixcam"

web/backend/api/channels.go

@@ -17,6 +17,7 @@ var channelCatalog = []channelCatalogItem{
 	{Name: "weixin", ConfigKey: "weixin"},
 	{Name: "telegram", ConfigKey: "telegram"},
 	{Name: "discord", ConfigKey: "discord"},
+	{Name: "imsg", ConfigKey: "imsg"},
 	{Name: "slack", ConfigKey: "slack"},
 	{Name: "feishu", ConfigKey: "feishu"},
 	{Name: "dingtalk", ConfigKey: "dingtalk"},

web/frontend/src/components/channels/channel-config-page.tsx

@@ -205,6 +205,7 @@ function getChannelDocSlug(channelName: string): string {
 
 const CHANNELS_WITHOUT_DOCS = new Set([
   "pico",
+  "imsg",
   "wecom",
   "matrix",
   "irc",

web/frontend/src/components/channels/channel-forms/generic-form.tsx

@@ -135,6 +135,7 @@ export function GenericForm({
       channels: t("channels.form.desc.channels"),
       request_caps: t("channels.form.desc.requestCaps"),
       max_base64_file_size_mib: t("channels.form.desc.maxBase64FileSizeMiB"),
+      iMessageCLIPath: t("channels.form.desc.iMessageCLIPath"),
     }
     return (
       descriptions[key] ??

web/frontend/src/hooks/use-sidebar-channels.ts

@@ -34,6 +34,7 @@ const CHANNEL_IMPORTANCE_TAIL = [
   "wecom",
   "dingtalk",
   "qq",
+  "imsg",
   "onebot",
   "matrix",
   "pico",
@@ -83,6 +84,7 @@ const CHANNEL_ICON_MAP: Record<
   onebot: IconRobot,
   pico: IconBrandChrome,
   irc: IconMessages,
+  imsg: IconMessages,
 }
 
 function asRecord(value: unknown): Record<string, unknown> {

web/frontend/src/i18n/locales/en.json

@@ -247,6 +247,7 @@
     "name": {
       "telegram": "Telegram",
       "discord": "Discord",
+      "imsg": "iMessage",
       "slack": "Slack",
       "feishu": "Feishu",
       "dingtalk": "DingTalk",
@@ -378,6 +379,7 @@
         "realName": "Displayed real name.",
         "channels": "IRC channels to join.",
         "requestCaps": "IRC capability list requested on connect.",
+        "iMessageCLIPath": "Path to iMessage CLI executable (default: imsg).",
         "maxBase64FileSizeMiB": "Maximum size in MiB for converting local files to base64 before upload. 0 means unlimited. Applies only to local files, not URL uploads.",
         "genericField": "Used to configure {{field}}."
       }

web/frontend/src/i18n/locales/zh.json

@@ -171,9 +171,8 @@
     "noDefaultHintPrefix": "尚未设置默认模型，点击",
     "noDefaultHintSuffix": "设为默认。",
     "status": {
-      "available": "可用",
-      "unconfigured": "未配置",
-      "unreachable": "服务不可达"
+      "configured": "已配置",
+      "unconfigured": "未配置"
     },
     "badge": {
       "default": "默认",
@@ -244,6 +243,10 @@
   },
   "channels": {
     "loadError": "加载频道列表失败",
+    "edit": "配置 {{name}}",
+    "status": {
+      "configured": "已配置"
+    },
     "name": {
       "telegram": "Telegram",
       "discord": "Discord",
@@ -263,6 +266,8 @@
       "weixin": "微信"
     },
     "weixin": {
+      "warningTitle": "测试阶段，请谨慎使用",
+      "warningDesc": "微信 Channel 当前仍处于测试阶段，存在封号风险。请仅在充分了解风险的前提下使用。",
       "bindTitle": "微信账号绑定",
       "bindDesc": "使用微信扫描二维码以绑定您的个人微信账号。",
       "bind": "绑定微信",
@@ -280,6 +285,8 @@
     "wecom": {
       "bindTitle": "企业微信绑定",
       "bindDesc": "使用企业微信扫描二维码以绑定您的 AI Bot。",
+      "enableDesc": "绑定后可在这里直接启用或停用频道。",
+      "enableBindFirst": "请先完成绑定，然后再启用频道。",
       "bind": "绑定企业微信",
       "rebind": "重新绑定",
       "bound": "企业微信已绑定",
@@ -315,12 +322,13 @@
       "allowOrigins": "允许来源域名",
       "allowOriginsPlaceholder": "例如 https://example.com, http://localhost:5173",
       "secretPlaceholder": "输入密钥",
-      "secretHintSet": "配置已保存，留空表示不修改"
+      "secretHintSet": "已设置密钥，留空表示不修改。"
     },
     "page": {
       "notFound": "不支持频道“{{name}}”。",
       "saveSuccess": "频道配置已保存。",
       "saveError": "保存频道配置失败",
+      "enabled": "已启用",
       "docLink": "配置文档",
       "enableLabel": "启用频道",
       "restartRequiredTitle": "需要重启服务",
@@ -331,55 +339,56 @@
         "token": "机器人访问令牌，用于连接平台 API",
         "botToken": "Bot Token，用于发送与接收消息",
         "appToken": "App Token，用于 Socket 模式连接。",
-        "appId": "应用唯一标识，用于平台鉴权",
-        "appSecret": "应用密钥，用于请求签名和鉴权",
-        "verificationToken": "事件回调验证令牌",
-        "encryptKey": "消息加密密钥，用于解密回调内容",
-        "baseUrl": "平台 API 地址，默认使用官方地址",
-        "proxy": "HTTP 代理地址，用于网络访问",
-        "mentionOnly": "在群聊中仅当明确提及时才响应",
-        "typingEnabled": "在生成回复时显示“正在输入”状态",
-        "placeholderEnabled": "在最终回复发送前，先发送临时占位消息",
-        "groupTriggerMentionOnly": "在群聊中仅当提及机器人时才响应",
-        "groupTriggerPrefixes": "群聊触发前缀，多个值用逗号分隔",
-        "isLark": "使用 Lark 国际版域名（open.larksuite.com）替代飞书域名（open.feishu.cn）",
-        "allowFrom": "允许访问的用户或群组 ID，多个值用逗号分隔",
-        "allowOrigins": "允许访问的来源域名，多个值用逗号分隔",
-        "wsUrl": "WebSocket 服务地址",
-        "reconnectInterval": "断线后的重连间隔（秒）",
-        "bridgeUrl": "桥接服务地址",
-        "sessionStorePath": "本地会话存储目录路径",
-        "useNative": "是否使用原生客户端模式连接",
-        "host": "服务监听主机地址",
-        "port": "服务监听端口",
-        "homeserver": "Matrix homeserver 地址",
-        "userId": "账号 ID",
-        "deviceId": "设备 ID",
-        "joinOnInvite": "收到邀请时是否自动加入房间",
-        "clientId": "应用客户端 ID，用于平台鉴权",
-        "corpId": "企业 ID",
-        "agentId": "企业应用 Agent ID",
-        "webhookUrl": "Webhook 完整地址",
-        "webhookHost": "Webhook 监听主机",
-        "webhookPort": "Webhook 监听端口",
-        "webhookPath": "Webhook 路径",
-        "replyTimeout": "回复超时时间（秒）",
-        "maxSteps": "最大步骤数",
-        "welcomeMessage": "新会话欢迎语内容",
-        "allowTokenQuery": "是否允许 URL Query 方式传递 Token",
-        "pingInterval": "连接心跳间隔（秒）",
-        "readTimeout": "读取超时时间（秒）",
-        "writeTimeout": "写入超时时间（秒）",
-        "maxConnections": "最大并发连接数",
-        "server": "IRC 服务器地址",
-        "tls": "是否启用 TLS 连接",
-        "nick": "机器人昵称",
-        "user": "IRC 用户名",
-        "realName": "显示名称",
-        "channels": "要加入的 IRC 频道列表",
-        "requestCaps": "连接时请求的 IRC 扩展能力列表",
-        "maxBase64FileSizeMiB": "本地文件转为 base64 上传的最大体积，单位 MiB；0 表示不限制，仅影响本地文件，不影响 URL 直传",
-        "genericField": "用于配置{{field}}"
+        "appId": "应用唯一标识，用于平台鉴权。",
+        "appSecret": "应用密钥，用于请求签名和鉴权。",
+        "verificationToken": "事件回调验证令牌。",
+        "encryptKey": "消息加密密钥，用于解密回调内容。",
+        "baseUrl": "平台 API 地址，默认使用官方地址。",
+        "proxy": "HTTP 代理地址，用于网络访问。",
+        "mentionOnly": "在群聊中仅当明确提及时才响应。",
+        "typingEnabled": "在生成回复时显示“正在输入”状态。",
+        "placeholderEnabled": "在最终回复发送前，先发送临时占位消息。",
+        "groupTriggerMentionOnly": "在群聊中仅当提及机器人时才响应。",
+        "groupTriggerPrefixes": "群聊触发前缀，多个值用逗号分隔。",
+        "isLark": "使用 Lark 国际版域名（open.larksuite.com）替代飞书域名（open.feishu.cn）。",
+        "allowFrom": "允许访问的用户或群组 ID，多个值用逗号分隔。",
+        "allowOrigins": "允许访问的来源域名，多个值用逗号分隔。",
+        "wsUrl": "WebSocket 服务地址。",
+        "reconnectInterval": "断线后的重连间隔（秒）。",
+        "bridgeUrl": "桥接服务地址。",
+        "sessionStorePath": "本地会话存储目录路径。",
+        "useNative": "是否使用原生客户端模式连接。",
+        "host": "服务监听主机地址。",
+        "port": "服务监听端口。",
+        "homeserver": "Matrix homeserver 地址。",
+        "userId": "账号 ID。",
+        "deviceId": "设备 ID。",
+        "joinOnInvite": "收到邀请时是否自动加入房间。",
+        "clientId": "应用客户端 ID，用于平台鉴权。",
+        "corpId": "企业 ID。",
+        "agentId": "企业应用 Agent ID。",
+        "webhookUrl": "Webhook 完整地址。",
+        "webhookHost": "Webhook 监听主机。",
+        "webhookPort": "Webhook 监听端口。",
+        "webhookPath": "Webhook 路径。",
+        "replyTimeout": "回复超时时间（秒）。",
+        "maxSteps": "最大步骤数。",
+        "welcomeMessage": "新会话欢迎语内容。",
+        "allowTokenQuery": "是否允许 URL Query 方式传递 Token。",
+        "pingInterval": "连接心跳间隔（秒）。",
+        "readTimeout": "读取超时时间（秒）。",
+        "writeTimeout": "写入超时时间（秒）。",
+        "maxConnections": "最大并发连接数。",
+        "server": "IRC 服务器地址。",
+        "tls": "是否启用 TLS 连接。",
+        "nick": "机器人昵称。",
+        "user": "IRC 用户名。",
+        "realName": "显示名称。",
+        "channels": "要加入的 IRC 频道列表。",
+        "requestCaps": "连接时请求的 IRC 扩展能力列表。",
+        "iMessageCLIPath": "iMessage CLI 可执行文件路径（默认：imsg）。",
+        "maxBase64FileSizeMiB": "本地文件转为 base64 上传的最大体积，单位 MiB；0 表示不限制，仅影响本地文件，不影响 URL 直传。",
+        "genericField": "用于配置{{field}}。"
       }
     },
     "validation": {
@@ -390,18 +399,11 @@
     "agent": {
       "load_error": "加载 Agent 支持信息失败。",
       "skills": {
+        "description": "技能会从工作区、PicoClaw 全局目录和内置目录中加载。",
         "empty": "当前没有可用技能。",
-        "install_success": "已安装 {{name}}。",
-        "install_error": "安装技能失败。",
-        "search_placeholder": "按名称、描述或技能源搜索",
-        "source_label": "类型",
-        "sort_label": "排序",
         "import": "导入技能",
         "import_success": "技能导入成功。",
         "import_error": "导入技能失败。",
-        "import_invalid_type": "仅支持导入 Markdown 或 ZIP 技能文件。",
-        "import_invalid_size": "技能文件大小不能超过 1 MB。",
-        "import_constraints": "支持导入最大 1 MB 的 Markdown 或 ZIP 文件",
         "view": "查看",
         "delete": "删除",
         "delete_title": "删除技能？",
@@ -411,78 +413,20 @@
         "delete_error": "删除技能失败。",
         "viewer_title": "技能内容",
         "viewer_description": "这里展示当前生效的 SKILL.md 内容。",
+        "loading_detail": "正在加载技能内容...",
         "load_detail_error": "加载技能内容失败。",
-        "no_description": "未提供描述。",
-        "no_results": "没有技能匹配当前筛选条件。",
-        "dropzone_title": "导入到工作区",
-        "dropzone_description": "将技能文件拖到这里，或从本地选择一个文件。",
-        "dropzone_label": "将技能文件拖到这里",
-        "dropzone_active": "松开即可导入该技能",
-        "dropzone_release": "导入后会自动规范化内容，并保存到工作区技能目录。",
-        "marketplace_title": "安装技能",
-        "marketplace_description": "搜索第三方技能源，并将技能安装到当前工作区",
-        "marketplace_search_placeholder": "搜索 github、docker、database 等技能",
-        "marketplace_search_action": "搜索",
-        "marketplace_search_status": "搜索状态",
-        "marketplace_install_status": "安装状态",
-        "marketplace_notice_title": "安全提示",
-        "marketplace_notice_body": "搜索结果中的 skills 属于第三方内容。安装前请先确认作者、页面 URL、说明文档，以及它要求执行的代码或使用的凭据是否可信。",
-        "marketplace_status_disabled": "当前未启用，请先在工具页启用对应工具。",
-        "marketplace_status_enable_hint": "请先在工具页启用相关工具。",
-        "marketplace_search_error": "搜索技能源失败。",
-        "marketplace_loading_results": "正在搜索技能...",
-        "marketplace_loading_more": "正在加载更多技能...",
-        "marketplace_results_title": "“{{query}}” 共找到 {{count}} 个结果",
-        "marketplace_results_hint": "搜索结果会安装到当前工作区。",
-        "marketplace_install_action": "安装",
-        "marketplace_installed": "已安装",
-        "marketplace_view_installed": "查看本地技能",
-        "marketplace_installed_hint": "该技能已在当前工作区中可用，名称为「{{name}}」。",
-        "marketplace_empty_results": "没有找到与“{{query}}”匹配的可安装技能。",
-        "marketplace_idle": "输入一个关键词，搜索可安装的第三方技能。",
-        "marketplace_unavailable": "当前无法使用技能搜索，请检查 Skills 相关工具配置。",
-        "sort": {
-          "name_asc": "名称（A-Z）",
-          "name_desc": "名称（Z-A）",
-          "source": "按类型"
-        },
-        "origin": {
-          "all": "全部类型",
-          "builtin": "内置",
-          "third_party": "第三方",
-          "manual": "手动导入"
-        },
-        "summary": {
-          "total": "技能总数"
-        },
-        "detail_tabs": {
-          "preview": "预览",
-          "raw": "原始内容",
-          "meta": "元数据"
-        },
-        "metadata": {
-          "name": "名称",
-          "description": "描述",
-          "registry": "来源平台",
-          "url": "链接地址",
-          "version": "已安装版本",
-          "lines": "行数",
-          "characters": "字符数"
-        }
+        "path": "技能路径",
+        "no_description": "未提供描述。"
       },
       "tools": {
-        "search_placeholder": "搜索工具...",
-        "no_results": "没有找到符合条件的工具",
-        "filter": {
-          "all": "所有状态",
-          "enabled": "已启用",
-          "disabled": "已禁用",
-          "blocked": "被阻塞"
-        },
+        "description": "这里展示每个 Agent 工具当前是已启用、已禁用，还是被依赖条件阻塞。",
         "empty": "当前没有可用工具。",
+        "enable": "启用",
+        "disable": "禁用",
         "enable_success": "工具已启用。",
         "disable_success": "工具已禁用。",
         "toggle_error": "更新工具状态失败。",
+        "config_key": "由 tools.{{key}} 控制",
         "status": {
           "enabled": "已启用",
           "disabled": "已禁用",
@@ -605,7 +549,6 @@
       "unsaved_changes": "您有未保存的更改。"
     },
     "logs": {
-      "log_level_error": "更新日志等级失败。",
       "clear": "清空日志",
       "empty": "等待日志中..."
     }