fix: 事故全链路修复 — 调度器合流 + 接管保护 + 补全去重 + 权限显示 + homeagent SSE 振荡
## 事故现场 用户选中补全里的「项目定位」→ 邮件投进另一条会话，界面显示的名字也不是 自己选的那个。授权页只显示 Agent 名，看不出哪个目录哪条线索。 ## 四处因果链 **① 调度器自己的 new_mail payload（起点）。** `notifyRecipients`（handler） 与 `SendCalendarMail`（scheduler）是两份代码。加 `platform_session_id` 时只改了 handler 那份 → 日历提醒投进接管会话时插件不知道是接管 → 另开一条新会话 → 命名同步冲掉接管会话的别名。 修法：抽出 `internal/notify` 包，唯一入口 `notify.Recipients`。 handler / scheduler / permission.go 都走它。新增字段时不存在「另一处忘了改」。 **② SyncSessionAlias 覆盖接管别名。** 别名是人从补全里选中的平台 slug， 任何平台命名同步都不该动它。加守卫 `platform_id <> ''` → 有绑定就返回当前值。 **③ SuggestSessionCandidates 按别名字符串去重。** 别名一被冲掉，同一条会话 出现两次（一次被冲的名字、一次镜像 slug），而另一条真实会话被吃掉。 改按 `platform_id` 去重。 mail 侧查 `s.platform_id`，镜像侧查 `platform_id`。 **④ FindOrCreateDefaultSession 不排除接管会话。** 日历提醒省略 session 位 → FindOrCreateDefaultSession 挑中人显式指定的接管会话。加 `platform_id = ''` 条件。 ## 权限页 **CreatePermissionMail 不写 from_workspace。** `from_workspace` 存空串 → 前端 `g.path && ...` 不渲染 → 人只看到光秃的 Agent 名，不知道哪个目录 哪条线索在请求权限。修法：INSERT 时从 sessions.workspace 取。 **SSE payload 缺 session_alias。** permission.go 的 SSE 不走 notify 包（决策人 不是地址解析出的参与方），但 payload 也要带 `session_alias` → 前端拼出 `pi@/home/program/agentmail.别名`，而不是光秃的 `pi`。 **mailGroups.ts：path ← session_workspace。** `from_workspace` 对 Agent 存的是 Agent 名（历史遗留），不能当路径用。PermissionList 显示完整三段地址 `agent@path.alias`。 ## NarrowStack z-index 窄屏日历的星期表头（`sticky top-0 z-10`）穿透到二级页面之上。覆盖层 auto z-index 输给 z-10 → 底层组件的层叠穿透到覆盖层。 修法：底层容器加 `isolate`（isolation: isolate），自成层叠上下文； 覆盖层加 `z-10`。只给覆盖层加 z-index 只能治当前一处，底层再写更大的 z-index 又会复现。 ## homeagent SSE 自激振荡 根因：五处缺陷叠加，SSE 每 60 秒断一次 → Gateway 全量重放 → 再断 → 再重放。 1. `p.client`（60s Timeout）跑 SSE 长连接 → 新增 `sseClient`（无超时） 2. `InjectInputSync` 在读循环里同步调用 → 改为 `go p.handleNewMail(evt)` 3. `lastEventID` 无条件赋值，Gateway 重放时发旧 ID → 单调递增 `sseMaxID` 4. 无邮件级去重 → 补 `deliveredMails map[string]bool` 5. 手动 `[]byte` 管理：每次 `buf[lineStart:]` 缩小 cap → 最终 len==cap → Read 零长切片 → 满速空转。改 `bufio.Reader`。 ## 清库 保留 jianf + 4 个 Agent 密钥 + 模型范围配置。清掉 mails/sessions/ calendar_events/attachments/agent_platform_sessions/relayed_mails/ permission_requests/rate_limits。测试数据已全部清零。 ## 测试 - gateway 7 包全过；repo + 7 例（adopt_alias_test.go） - web 182 例（mailGroups 新增 session_workspace 断言） - 前端构建通过
This commit is contained in:
@ -1,6 +1,7 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
@ -73,6 +74,22 @@ type Plugin struct {
|
||||
|
||||
// B-1.6 补拉状态：首个成功心跳后只补一次
|
||||
catchupDone bool
|
||||
|
||||
// ─── SSE 专用 ───
|
||||
|
||||
// SSE 需要一个不设 Timeout 的 HTTP client：原来 p.client（60s Timeout）
|
||||
// 跑 SSE 长连接，每 60 秒自己掐断自己。之后 lastEventID 回退 → 重放 →
|
||||
// 又阻塞 → 又超时 —— 自激振荡。这个 client 只给 readSSE 用。
|
||||
sseClient *http.Client
|
||||
|
||||
// B-7.3 邮件级去重：SSE 重放会重发同一批事件，没有这层去重
|
||||
// 每封邮件会被注入 agent 两遍。契约 B-7.3 要求：每封只注入一次。
|
||||
deliveredMails map[string]bool
|
||||
|
||||
// 单调递增的 last-seen-ID：被重放的旧事件不会让它回退。
|
||||
// 原来直接赋值（p.lastEventID = eid），Gateway 重放时发旧 ID，
|
||||
// 于是 lastEventID 从 123 退回 116 → 下次重连又报 116 → 又重放。
|
||||
sseMaxID int64
|
||||
}
|
||||
|
||||
// ─── B-1.1 密钥解析与本地生成 ───
|
||||
@ -146,14 +163,16 @@ func NewPluginFactory(name string, config map[string]interface{}) (sdk.Plugin, e
|
||||
}
|
||||
|
||||
return &Plugin{
|
||||
name: name,
|
||||
agentName: agentName,
|
||||
gwURL: strings.TrimRight(gw, "/"),
|
||||
key: "", // Start() 里解析
|
||||
keyFile: "",
|
||||
client: &http.Client{Timeout: 60 * time.Second},
|
||||
stopCh: make(chan struct{}),
|
||||
explicitSends: make(map[string]time.Time),
|
||||
name: name,
|
||||
agentName: agentName,
|
||||
gwURL: strings.TrimRight(gw, "/"),
|
||||
key: "", // Start() 里解析
|
||||
keyFile: "",
|
||||
client: &http.Client{Timeout: 60 * time.Second},
|
||||
sseClient: &http.Client{}, // 无超时：SSE 是长连接
|
||||
stopCh: make(chan struct{}),
|
||||
deliveredMails: make(map[string]bool),
|
||||
explicitSends: make(map[string]time.Time),
|
||||
}, nil
|
||||
}
|
||||
|
||||
@ -550,14 +569,19 @@ func (p *Plugin) readSSE() error {
|
||||
}
|
||||
req.Header.Set("Authorization", "Bearer "+p.key)
|
||||
|
||||
// W-4：断线期间的事件会丢，带上 Last-Event-ID 可以让 Gateway 从断点补发
|
||||
// W-4：断线期间的事件会丢，带上 Last-Event-ID 可以让 Gateway 从断点补发。
|
||||
// 只上报比当前记录的更大的 ID：Gateway 重放时发的是事件原本的 ID，
|
||||
// 如果无条件赋值，lastEventID 会从 123 退回 116 → 下次重连又报 116 →
|
||||
// 又重放 —— 自激振荡的放大器。
|
||||
p.sseMu.Lock()
|
||||
if p.lastEventID != "" {
|
||||
req.Header.Set("Last-Event-ID", p.lastEventID)
|
||||
}
|
||||
p.sseMu.Unlock()
|
||||
|
||||
resp, err := p.client.Do(req)
|
||||
// sseClient 无 Timeout：p.client 有 60s Timeout，SSE 是长连接，
|
||||
// 每 60 秒自己掐断自己 → 重放 → 阻塞 → 超时 → 重放。
|
||||
resp, err := p.sseClient.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@ -569,8 +593,11 @@ func (p *Plugin) readSSE() error {
|
||||
|
||||
log.Printf("[homeagent-mail-bridge] SSE 已连接")
|
||||
|
||||
buf := make([]byte, 0, 4096)
|
||||
lineStart := 0
|
||||
// bufio.Reader 解决原来手动管理 []byte 的两个问题：
|
||||
// 1. 每次 buf = buf[lineStart:] 让 cap 缩小，几轮之后 len==cap，
|
||||
// Read 拿到零长切片 → (0, nil) → 满速空转
|
||||
// 2. 手写的 line 分割逻辑有边界条件（跨次 Read 的半行处理）
|
||||
br := bufio.NewReader(resp.Body)
|
||||
for {
|
||||
select {
|
||||
case <-p.stopCh:
|
||||
@ -578,39 +605,43 @@ func (p *Plugin) readSSE() error {
|
||||
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)]
|
||||
line, err := br.ReadString('\n')
|
||||
if line != "" {
|
||||
p.parseSSELine(strings.TrimRight(line, "\n"))
|
||||
}
|
||||
if err != nil {
|
||||
if err != io.EOF {
|
||||
return err
|
||||
if err == io.EOF {
|
||||
return nil
|
||||
}
|
||||
return nil
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// parseSSEID 把 "id: 123" 格式的事件 ID 解析成整数。
|
||||
// 解析失败返回 0（大于 0 的 ID 才会被接受），保证不会误清状态。
|
||||
func parseSSEID(raw string) int64 {
|
||||
var id int64
|
||||
for _, c := range raw {
|
||||
if c >= '0' && c <= '9' {
|
||||
id = id*10 + int64(c-'0')
|
||||
}
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
func (p *Plugin) parseSSELine(line string) {
|
||||
// W-4：记录 Last-Event-ID
|
||||
// W-4：记录 Last-Event-ID。只向前推进，不回退。
|
||||
// Gateway 重放旧事件时发的是旧 ID，无条件赋值会让 lastEventID
|
||||
// 从 123 退回到 116 → 下次重连报 116 → 又重放 → 振荡。
|
||||
if strings.HasPrefix(line, "id: ") {
|
||||
eid := strings.TrimPrefix(line, "id: ")
|
||||
id := parseSSEID(eid)
|
||||
p.sseMu.Lock()
|
||||
p.lastEventID = eid
|
||||
if id > p.sseMaxID {
|
||||
p.sseMaxID = id
|
||||
p.lastEventID = eid
|
||||
}
|
||||
p.sseMu.Unlock()
|
||||
return
|
||||
}
|
||||
@ -647,7 +678,22 @@ func (p *Plugin) parseSSELine(line string) {
|
||||
}
|
||||
|
||||
if evt.MailType == "normal" {
|
||||
p.handleNewMail(evt)
|
||||
// B-7.3：去重。SSE 重放时同一封邮件会再出现，没有这层
|
||||
// 每封邮件会被注入 agent 两遍（实测 21 次超时 → 21 次重放）。
|
||||
p.sseMu.Lock()
|
||||
if p.deliveredMails[evt.MailID] {
|
||||
p.sseMu.Unlock()
|
||||
return
|
||||
}
|
||||
p.deliveredMails[evt.MailID] = true
|
||||
p.sseMu.Unlock()
|
||||
|
||||
// InjectInputSync 会阻塞几十秒（查日志、调工具、转发 QQ），
|
||||
// 而它跑在 readSSE 的读循环里 —— 循环卡住期间 SSE 事件积压在
|
||||
// TCP 缓冲区，卡到超时断线重连后 Gateway 全部重放一遍。
|
||||
// 把处理丢到独立 goroutine：parseSSELine 立刻返回，读循环继续。
|
||||
// homeagent 是单事件循环，InjectInputSync 自己会排队。
|
||||
go p.handleNewMail(evt)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user