MailUI4Agents/plugins/homeagent-mail-bridge/plugin.go

package main

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"strings"
	"sync"
	"time"

	"gitcode.com/JianFeeeee/homeagent-sdk/sdk"
)

const (
	defaultGateway  = "http://127.0.0.1:8180"
	heartbeatIntval = 25 * time.Second
	sseRetry        = 3 * time.Second
	mailPollIntval  = 15 * time.Second
)

// Plugin 实现 sdk.Plugin。
//
// TrueAgent 是单一常驻 agent 事件循环，没有「每个对话一个 session」的概念。
// 因此本插件不做会话映射 —— 所有邮件注入同一个事件循环，像 QQ 插件一样。
// 邮件的 context 完全靠中断消息的文本传递，不靠 platform_sessions 上报。
type Plugin struct {
	name         string
	sdk          *sdk.PluginSDK
	gwURL        string
	key          string
	client       *http.Client
	stopCh       chan struct{}
	stopOnce     sync.Once
}

func NewPluginFactory(name string, config map[string]interface{}) (sdk.Plugin, error) {
	gw := ""
	key := ""
	if v, ok := config["gateway_url"].(string); ok {
		gw = v
	}
	if v, ok := config["gateway_key"].(string); ok {
		key = v
	}
	if gw == "" {
		gw = os.Getenv("AGENTMAIL_GATEWAY_URL")
	}
	if gw == "" {
		gw = defaultGateway
	}
	if key == "" {
		key = os.Getenv("AGENTMAIL_AGENT_KEY")
	}
	return &Plugin{
		name:   name,
		gwURL:  strings.TrimRight(gw, "/"),
		key:    key,
		client: &http.Client{Timeout: 30 * time.Second},
		stopCh: make(chan struct{}),
	}, nil
}

func (p *Plugin) Name() string { return p.name }

func (p *Plugin) Start(s *sdk.PluginSDK) error {
	p.sdk = s
	s.SetAutoRestart(true)

	// 注册读取工具
	s.RegisterTool("read_inbox", sdk.ToolDef{
		Name:        "read_inbox",
		Description: "查阅收件箱中的邮件。收到新邮件通知后应立即调用此工具。每封含 mail_id、发件人、主题、正文与附件清单。",
		Parameters: map[string]interface{}{
			"type": "object",
			"properties": map[string]interface{}{
				"status": map[string]interface{}{
					"type":        "string",
					"description": "过滤条件 unread|all，默认 unread",
				},
				"limit": map[string]interface{}{
					"type":        "number",
					"description": "返回数量，默认 5",
				},
			},
		},
	}, p.handleReadInbox)

	// 注册发信工具
	s.RegisterTool("send_mail", sdk.ToolDef{
		Name:        "send_mail",
		Description: "发送邮件。三维地址 name@path.session。回复来信请传 reply_to。",
		Parameters: map[string]interface{}{
			"type": "object",
			"properties": map[string]interface{}{
				"to":       map[string]interface{}{"type": "string", "description": "收件人三维地址"},
				"subject":  map[string]interface{}{"type": "string", "description": "邮件主题"},
				"body":     map[string]interface{}{"type": "string", "description": "邮件正文（Markdown）"},
				"cc":       map[string]interface{}{"type": "string", "description": "抄送"},
				"reply_to": map[string]interface{}{"type": "string", "description": "回复某封邮件时传其 mail_id"},
			},
			"required": []string{"to", "subject", "body"},
		},
	}, p.handleSendMail)

	// 注册输出通道 —— agent 可以主动调 output_send__homeagent 发信
	s.RegisterOutputChannel("homeagent", sdk.CapText|sdk.CapFile,
		"发送邮件。meta JSON 格式：{to, subject, reply_to}，type: text",
		sdk.ChannelDef{}, p.handleOutputChannel)

	log.Printf("[homeagent-mail-bridge] 注册完成，等待 Gateway SSE")

	// 启动心跳 + SSE（后台 goroutine）
	go p.heartbeatLoop()
	go p.sseLoop()

	return nil
}

func (p *Plugin) Stop() error {
	p.stopOnce.Do(func() { close(p.stopCh) })
	return nil
}

// ─── 心跳 ───

func (p *Plugin) heartbeatLoop() {
	// 首次注册
	if err := p.register(); err != nil {
		log.Printf("[homeagent-mail-bridge] 注册失败: %v", err)
	}
	ticker := time.NewTicker(heartbeatIntval)
	defer ticker.Stop()
	for {
		select {
		case <-p.stopCh:
			return
		case <-ticker.C:
			if err := p.heartbeat(); err != nil {
				log.Printf("[homeagent-mail-bridge] 心跳失败: %v", err)
			}
		}
	}
}

func (p *Plugin) register() error {
	body := map[string]interface{}{
		"name":       p.name,
		"platform":   "homeagent",
		"workspaces": []interface{}{},
	}
	return p.post("/agent/register", body, nil)
}

func (p *Plugin) heartbeat() error {
	return p.post("/agent/heartbeat", map[string]interface{}{}, nil)
}

// ─── SSE ───

func (p *Plugin) sseLoop() {
	for {
		select {
		case <-p.stopCh:
			return
		default:
		}
		if err := p.readSSE(); err != nil {
			log.Printf("[homeagent-mail-bridge] SSE 断开: %v，%v 后重连", err, sseRetry)
			time.Sleep(sseRetry)
		}
	}
}

func (p *Plugin) readSSE() error {
	req, err := http.NewRequest("GET", p.gwURL+"/api/v1/events/stream", nil)
	if err != nil {
		return err
	}
	req.Header.Set("Authorization", "Bearer "+p.key)

	resp, err := p.client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	if resp.StatusCode != 200 {
		return fmt.Errorf("SSE HTTP %d", resp.StatusCode)
	}

	log.Printf("[homeagent-mail-bridge] SSE 已连接")

	buf := make([]byte, 0, 4096)
	lineStart := 0
	for {
		select {
		case <-p.stopCh:
			return nil
		default:
		}

		n, err := resp.Body.Read(buf[len(buf):cap(buf)])
		if n > 0 {
			buf = buf[:len(buf)+n]
			// 处理完整行
			for {
				i := bytes.IndexByte(buf[lineStart:], '\n')
				if i < 0 {
					break
				}
				line := string(buf[lineStart : lineStart+i])
				lineStart += i + 1
				p.parseSSELine(line)
			}
			if lineStart > 0 {
				buf = buf[lineStart:]
				lineStart = 0
			}
			buf = buf[:len(buf)]
		}
		if err != nil {
			if err != io.EOF {
				return err
			}
			return nil
		}
	}
}

func (p *Plugin) parseSSELine(line string) {
	// SSE 格式：event: xxx\ndata: {...}\n\n
	if !strings.HasPrefix(line, "data: ") {
		return
	}
	raw := strings.TrimPrefix(line, "data: ")
	if raw == "" || raw == "{}" {
		return
	}

	var evt struct {
		MailID    string `json:"mail_id"`
		SessionID string `json:"session_id"`
		FromName  string `json:"from_name"`
		Subject   string `json:"subject"`
		MailType  string `json:"mail_type"`
		Role      string `json:"role"`
		Workspace string `json:"to_workspace"`
		Alias     string `json:"session_alias"`
		ReplyAddr string `json:"reply_address"`
	}
	if err := json.Unmarshal([]byte(raw), &evt); err != nil {
		return
	}

	if evt.MailID == "" {
		return
	}

	// 权限决策回复（MUST）
	if evt.MailType == "permission_decision" {
		p.handlePermissionDecision(evt)
		return
	}

	// 新邮件：注入 agent 事件循环
	if evt.MailType == "normal" {
		p.handleNewMail(evt)
	}
}

// ─── 输出通道（agent 主动发信）───

func (p *Plugin) handleOutputChannel(args map[string]interface{}) (interface{}, error) {
	payload, _ := args["payload"].(string)
	meta, _ := args["meta"].(string)
	if payload == "" {
		return nil, fmt.Errorf("payload 不能为空")
	}

	var m struct {
		To      string `json:"to"`
		Subject string `json:"subject"`
		ReplyTo string `json:"reply_to"`
	}
	if meta != "" {
		json.Unmarshal([]byte(meta), &m)
	}
	if m.To == "" {
		return nil, fmt.Errorf("meta 中需要 to 字段")
	}

	if err := p.sendMail(m.To, m.Subject, payload, m.ReplyTo); err != nil {
		return nil, err
	}
	return map[string]interface{}{"status": "sent"}, nil
}

func (p *Plugin) sendMail(to, subject, body, replyTo string) error {
	payload := map[string]interface{}{
		"to":      to,
		"subject": subject,
		"body":    body,
	}
	if replyTo != "" {
		payload["reply_to"] = replyTo
	}
	return p.post("/mail/send", payload, nil)
}

// ─── 新邮件处理 ───

func (p *Plugin) handleNewMail(evt struct {
	MailID    string `json:"mail_id"`
	SessionID string `json:"session_id"`
	FromName  string `json:"from_name"`
	Subject   string `json:"subject"`
	MailType  string `json:"mail_type"`
	Role      string `json:"role"`
	Workspace string `json:"to_workspace"`
	Alias     string `json:"session_alias"`
	ReplyAddr string `json:"reply_address"`
}) {
	// 记录回信目标 —— StageBeforeOutput 会检查是否由邮件触发，
	// StageAfterOutput 用这个地址把最终回复发回去。
	// 构建中断消息
	prompt := fmt.Sprintf(
		"你收到一封新邮件（AgentMail）。\n\n"+
			"发件人：%s\n"+
			"主题：%s\n"+
			"邮件 ID：%s\n"+
			"身份：你是 %s\n\n"+
			"请先调用 read_inbox 读取完整正文，然后处理其中的请求。\n\n"+
			"**回信不用你自己发**：你把本轮工作做完、把结论说出来就行，\n"+
			"插件会在这一轮结束时自动把你最后那段话作为回信发回给 %s（不消耗你的发信配额）。\n"+
			"只有在需要主动联系其他人、或要带附件时才调用 send_mail。",
		evt.FromName, evt.Subject, evt.MailID, p.name, evt.FromName,
	)

	// 非阻塞注入 —— TrueAgent 的事件循环会处理
	// InjectInputSync 阻塞等待 agent 处理完毕，返回最终回复文本。
	// 这比 stage handler 更可靠：TrueAgent 的事件循环保证不会重入，
	// 而 stage handler 的触发时机依赖 ToolCalls 的状态快照，实际测试中
	// before_output 看不到 read_inbox 调用（它在 turn 中间就被清掉了）。
	reply := p.sdk.InjectInputSync(p.name, p.name, prompt)
	if reply == "" {
		log.Printf("[homeagent-mail-bridge] 已注入新邮件 %s（来自 %s：%s），agent 无回复", evt.MailID[:8], evt.FromName, evt.Subject)
		return
	}
	// 自动把 agent 的回复发回给发件人
	if err := p.sendMail(evt.FromName, "Re: "+evt.Subject, reply, evt.MailID); err != nil {
		log.Printf("[homeagent-mail-bridge] 自动回信失败: %v", err)
	} else {
		log.Printf("[homeagent-mail-bridge] 已自动回信给 %s（%d 字）", evt.FromName, len(reply))
	}
}

// ─── 权限决策 ───

func (p *Plugin) handlePermissionDecision(evt struct {
	MailID    string `json:"mail_id"`
	SessionID string `json:"session_id"`
	FromName  string `json:"from_name"`
	Subject   string `json:"subject"`
	MailType  string `json:"mail_type"`
	Role      string `json:"role"`
	Workspace string `json:"to_workspace"`
	Alias     string `json:"session_alias"`
	ReplyAddr string `json:"reply_address"`
}) {
	prompt := fmt.Sprintf(
		"你之前发起的权限请求已有结论：%s（决策人：%s）。请据此继续。",
		evt.Subject, evt.FromName,
	)
	p.sdk.InjectText(p.name, p.name, prompt)
}

// ─── 工具实现 ───

func (p *Plugin) handleReadInbox(args map[string]interface{}) (interface{}, error) {
	status := "unread"
	if v, ok := args["status"].(string); ok && v != "" {
		status = v
	}
	limit := 5
	if v, ok := args["limit"].(float64); ok && v > 0 {
		limit = int(v)
	}

	url := fmt.Sprintf("%s/api/v1/mail/inbox?status=%s&limit=%d", p.gwURL, status, limit)
	var result map[string]interface{}
	if err := p.get(url, &result); err != nil {
		return nil, err
	}

	// 渲染成模型可读的文本
	mails, _ := result["mails"].([]interface{})
	if len(mails) == 0 {
		return map[string]interface{}{"content": []map[string]interface{}{{"type": "text", "text": "收件箱为空。"}}}, nil
	}

	var sb strings.Builder
	for i, m := range mails {
		mail, _ := m.(map[string]interface{})
		if mail == nil {
			continue
		}
		from, _ := mail["from_name"].(string)
		subj, _ := mail["subject"].(string)
		mid, _ := mail["mail_id"].(string)
		body, _ := mail["body"].(string)
		if body == "" {
			body, _ = mail["body_preview"].(string)
		}
		alias, _ := mail["session_alias"].(string)

		fmt.Fprintf(&sb, "[%d] %s: %s\n邮件 ID: %s\n会话: #%s\n",
			i+1, from, subj, mid, alias)

		// 抄送
		if ccList, ok := mail["cc_list"].([]interface{}); ok && len(ccList) > 0 {
			names := make([]string, 0, len(ccList))
			for _, c := range ccList {
				if cc, ok := c.(map[string]interface{}); ok {
					if n, ok := cc["name"].(string); ok {
						names = append(names, n)
					}
				}
			}
			if len(names) > 0 {
				fmt.Fprintf(&sb, "抄送: %s\n", strings.Join(names, "、"))
			}
		}

		// 附件
		if atts, ok := mail["attachments"].([]interface{}); ok && len(atts) > 0 {
			fmt.Fprintf(&sb, "附件:\n")
			for _, a := range atts {
				if att, ok := a.(map[string]interface{}); ok {
					fn, _ := att["filename"].(string)
					sz, _ := att["size_bytes"].(float64)
					aid, _ := att["attachment_id"].(string)
					fmt.Fprintf(&sb, "  - %s (%.1fKB, id=%s)\n", fn, sz/1024, aid)
				}
			}
		}

		if body != "" {
			if len(body) > 1000 {
				body = body[:1000] + "..."
			}
			fmt.Fprintf(&sb, "内容: %s\n", body)
		}
		sb.WriteString("\n")
	}

	// 标记已读
	ids := make([]string, 0, len(mails))
	for _, m := range mails {
		if mail, ok := m.(map[string]interface{}); ok {
			if id, ok := mail["mail_id"].(string); ok {
				ids = append(ids, id)
			}
		}
	}
	if len(ids) > 0 {
		go p.markRead(ids)
	}

	return map[string]interface{}{
		"content": []map[string]interface{}{{"type": "text", "text": sb.String()}},
	}, nil
}

func (p *Plugin) handleSendMail(args map[string]interface{}) (interface{}, error) {
	to, _ := args["to"].(string)
	subj, _ := args["subject"].(string)
	body, _ := args["body"].(string)
	cc, _ := args["cc"].(string)
	replyTo, _ := args["reply_to"].(string)

	if to == "" || subj == "" || body == "" {
		return nil, fmt.Errorf("缺少必填字段：to, subject, body")
	}

	payload := map[string]interface{}{
		"to":       to,
		"subject":  subj,
		"body":     body,
	}
	if cc != "" {
		payload["cc"] = cc
	}
	if replyTo != "" {
		payload["reply_to"] = replyTo
	}

	var result map[string]interface{}
	if err := p.post("/mail/send", payload, &result); err != nil {
		return nil, err
	}

	mid, _ := result["mail_id"].(string)
	sid, _ := result["session_id"].(string)
	text := fmt.Sprintf("邮件已发送（ID: %s，Session: %s）", mid, sid)
	if budget, ok := result["budget_remaining"].(float64); ok {
		text += fmt.Sprintf("。本任务剩余 %.0f 个来回", budget)
	}

	return map[string]interface{}{
		"content": []map[string]interface{}{{"type": "text", "text": text}},
	}, nil
}

// ─── HTTP 辅助 ───

func (p *Plugin) get(url string, out interface{}) error {
	req, err := http.NewRequest("GET", url, nil)
	if err != nil {
		return err
	}
	req.Header.Set("Authorization", "Bearer "+p.key)

	resp, err := p.client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	if resp.StatusCode >= 400 {
		body, _ := io.ReadAll(resp.Body)
		return fmt.Errorf("HTTP %d: %s", resp.StatusCode, string(body))
	}
	return json.NewDecoder(resp.Body).Decode(out)
}

func (p *Plugin) post(path string, payload interface{}, out interface{}) error {
	data, err := json.Marshal(payload)
	if err != nil {
		return err
	}

	req, err := http.NewRequest("POST", p.gwURL+"/api/v1"+path, bytes.NewReader(data))
	if err != nil {
		return err
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+p.key)

	resp, err := p.client.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	if resp.StatusCode >= 400 {
		body, _ := io.ReadAll(resp.Body)
		return fmt.Errorf("POST %s HTTP %d: %s", path, resp.StatusCode, string(body))
	}
	if out != nil {
		return json.NewDecoder(resp.Body).Decode(out)
	}
	return nil
}

func (p *Plugin) markRead(ids []string) {
	data, _ := json.Marshal(map[string]interface{}{"mail_ids": ids})
	req, err := http.NewRequest("POST", p.gwURL+"/api/v1/mail/read", bytes.NewReader(data))
	if err != nil {
		return
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+p.key)
	p.client.Do(req)
}