diff --git a/internal/channel/adapter/delivery.go b/internal/channel/adapter/delivery.go new file mode 100644 index 0000000..ee44f92 --- /dev/null +++ b/internal/channel/adapter/delivery.go @@ -0,0 +1,90 @@ +package adapter + +import "errors" + +// 本文件给**出站投递失败**一个稳定的原因码。此前一次失败只有自由文本(真机日志: +// `weixin_ilink: proactive push outside 24h window`),调用方只能把整串原文塞进日志与 +// 记录 —— 于是「她的话没送到微信」在面板上只表现为一句人读的英文,既没法稳定地告诉 +// 主人「哪个渠道、因为哪条平台规则」,也没法按原因聚合。 +// +// 口径与 internal/runtime/voice 的 `Category(err)` 同族(那处给语音转码失败分类): +// **产生错误的那一层**声明码值,读取侧只调一个访问函数;错误**没有自报**更具体的 +// 原因码时一律落到兜底码 CategoryFailed —— 绝不替它编一个更精确的码。 +// +// 为什么放在 adapter 契约层:能见到真实错误的层(plugin 适配器)与要记录它的层 +// (channelbridge / backthink)之间只共享本包。契约层是唯一不需要任何一方反向依赖 +// 具体插件的落点。范式与 SystemNoticeNormalizer 逐字同族:**可选接口 + 调用方探测**。 + +// DeliveryCategory 是出站投递失败的**稳定原因码**(snake_case,可直接进日志字段、 +// 记录字段与审计文案)。机器读它,人读文案 —— 所以文案会变,码不变。 +type DeliveryCategory string + +// 契约层自带的四个码。平台特有的码(如 weixin-ilink 的 proactive_window_closed) +// 由各插件在自己的包内声明并自报,本包不枚举它们。 +const ( + // CategoryNotConnected 表示适配器当前没有可用会话(ErrNotConnected)。 + CategoryNotConnected DeliveryCategory = "not_connected" + // CategoryUnsupported 表示该能力/操作不被平台支持(ErrUnsupported)。 + CategoryUnsupported DeliveryCategory = "unsupported" + // CategoryClosed 表示适配器已关闭(ErrClosed)。 + CategoryClosed DeliveryCategory = "closed" + // CategoryFailed 是**兜底**码:错误没有自报更具体的原因码时使用。它如实表示 + // 「只知道失败了,不知道平台规则层面的原因」,既不是成功,也不是猜测。 + CategoryFailed DeliveryCategory = "failed" +) + +// DeliveryFailure 是投递错误**可选**的自我描述(适配器按需实现,调用方用 +// errors.As 探测)。实现必须无副作用:不做 I/O、不改状态。 +type DeliveryFailure interface { + // DeliveryCategory 返回该错误的稳定原因码。返回空串表示「不自报」,读取侧 + // 退到共享哨兵判定与兜底码。 + DeliveryCategory() DeliveryCategory + // DeliveryChannel 返回出错渠道的平台名(例如 "weixin-ilink")。空串表示 + // **不知道**——绝不填一个猜出来的平台名。 + DeliveryChannel() string +} + +// DeliveryCategoryOf 返回 err 的稳定原因码:err 自报则用自报值,否则按契约层 +// 哨兵判定,最后落到 CategoryFailed。err 为 nil 时返回空串 —— 「没有失败」不是 +// 一种失败码。 +// +// errors.As 沿 %w 链走,因此中间层的包装(channelbridge 的 +// `send bubble #1 of 8: …`、`send proactive via "weixin-ilink": …`)都不会把 +// 自报的码藏起来。 +func DeliveryCategoryOf(err error) DeliveryCategory { + if err == nil { + return "" + } + if f := deliveryFailureOf(err); f != nil { + if c := f.DeliveryCategory(); c != "" { + return c + } + } + switch { + case errors.Is(err, ErrNotConnected): + return CategoryNotConnected + case errors.Is(err, ErrUnsupported): + return CategoryUnsupported + case errors.Is(err, ErrClosed): + return CategoryClosed + default: + return CategoryFailed + } +} + +// DeliveryChannelOf 返回 err 自报的出错渠道;未自报时返回空串(= 未知)。 +func DeliveryChannelOf(err error) string { + if f := deliveryFailureOf(err); f != nil { + return f.DeliveryChannel() + } + return "" +} + +// deliveryFailureOf 在错误链上找出第一个自描述失败;没有则返回 nil。 +func deliveryFailureOf(err error) DeliveryFailure { + var f DeliveryFailure + if errors.As(err, &f) { + return f + } + return nil +} diff --git a/internal/channel/adapter/delivery_test.go b/internal/channel/adapter/delivery_test.go new file mode 100644 index 0000000..8edae64 --- /dev/null +++ b/internal/channel/adapter/delivery_test.go @@ -0,0 +1,74 @@ +package adapter + +import ( + "errors" + "fmt" + "testing" +) + +// selfDescribedFailure 是自描述投递失败的替身(适配器实现 DeliveryFailure 的最小形状)。 +type selfDescribedFailure struct { + category DeliveryCategory + channel string + text string +} + +func (e *selfDescribedFailure) Error() string { return e.text } + +func (e *selfDescribedFailure) DeliveryCategory() DeliveryCategory { return e.category } + +func (e *selfDescribedFailure) DeliveryChannel() string { return e.channel } + +// TestDeliveryCategoryOf 锁定原因码口径:自报优先、契约层哨兵次之、兜底 +// CategoryFailed —— 绝不替一个没说清楚原因的错误编一个更精确的码。 +func TestDeliveryCategoryOf(t *testing.T) { + t.Parallel() + window := &selfDescribedFailure{ + category: "proactive_window_closed", + channel: "weixin-ilink", + text: "weixin_ilink: proactive push outside 24h window", + } + tests := []struct { + name string + err error + want DeliveryCategory + }{ + {name: "没有失败就没有码", err: nil, want: ""}, + {name: "适配器自报的码被采信", err: window, want: "proactive_window_closed"}, + {name: "多气泡包装不藏自报码", err: fmt.Errorf("channelbridge: send bubble #1 of 2: %w", window), want: "proactive_window_closed"}, + {name: "桥接器包装不藏自报码", err: fmt.Errorf("channelbridge: send proactive via %q: %w", "weixin-ilink", window), want: "proactive_window_closed"}, + {name: "未连接", err: ErrNotConnected, want: CategoryNotConnected}, + {name: "不支持", err: fmt.Errorf("adapter: send: %w", ErrUnsupported), want: CategoryUnsupported}, + {name: "已关闭", err: ErrClosed, want: CategoryClosed}, + {name: "无自报的裸错误落兜底", err: errors.New("boom"), want: CategoryFailed}, + {name: "自报空码落兜底(不编更精确的码)", err: &selfDescribedFailure{text: "boom"}, want: CategoryFailed}, + } + for _, tc := range tests { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if got := DeliveryCategoryOf(tc.err); got != tc.want { + t.Fatalf("DeliveryCategoryOf(%v) = %q, want %q", tc.err, got, tc.want) + } + }) + } +} + +// TestDeliveryChannelOf 锁定渠道自报口径:只有自报才有渠道名,未自报是**空串** +// (= 不知道),绝不替它猜一个平台名。 +func TestDeliveryChannelOf(t *testing.T) { + t.Parallel() + window := &selfDescribedFailure{category: "proactive_window_closed", channel: "weixin-ilink", text: "closed"} + if got := DeliveryChannelOf(fmt.Errorf("channelbridge: send bubble #1 of 2: %w", window)); got != "weixin-ilink" { + t.Fatalf("DeliveryChannelOf(wrapped) = %q, want weixin-ilink", got) + } + if got := DeliveryChannelOf(fmt.Errorf("channelbridge: no connected owner IM channel for proactive push")); got != "" { + t.Fatalf("DeliveryChannelOf(unreported) = %q, want empty (unknown)", got) + } + if got := DeliveryChannelOf(nil); got != "" { + t.Fatalf("DeliveryChannelOf(nil) = %q, want empty", got) + } + if got := DeliveryChannelOf(ErrNotConnected); got != "" { + t.Fatalf("DeliveryChannelOf(ErrNotConnected) = %q, want empty", got) + } +} diff --git a/internal/services/backthink/backthink.go b/internal/services/backthink/backthink.go index 5d46183..4fcfe49 100644 --- a/internal/services/backthink/backthink.go +++ b/internal/services/backthink/backthink.go @@ -170,9 +170,6 @@ func New(cfg Config, deps Deps) (*Backthink, error) { gate = newRateGate(cfg.PushLimit, cfg.PushWindow, cfg.PushCooldown, cfg.KindWindow, log) } pusher := deps.Pusher - if pusher == nil { - pusher = newPusher(deps.Active, deps.IM, gate, log).withExporter(deps.Exporter) - } now := deps.Now if now == nil { @@ -183,7 +180,7 @@ func New(cfg Config, deps Deps) (*Backthink, error) { sessionID = defaultSessionID } - return &Backthink{ + b := &Backthink{ cfg: cfg, deps: deps, log: log, @@ -195,7 +192,13 @@ func New(cfg Config, deps Deps) (*Backthink, error) { // 闸只认**已 normalize** 的旋钮:未回落默认的 ≤0 会被滑窗读成「无规则」 // (恒放行且不记录),那就把「≤0 回落默认」静默变成「无上限」(G3)。 selfWake: newSelfWakeGate(cfg.SelfWakeLimit, cfg.SelfWakeWindow), - }, nil + } + // 未注入自定义推送器时,默认推送器在结构体构造**之后**组装:它要挂上 + // b.auditEvent(读取当前审计出口的回调),而该回调需要 b 本身。 + if pusher == nil { + b.rebuildPusher() + } + return b, nil } // Start 启动后台工作:心跳定时器,以及(配置了 Bus 时) diff --git a/internal/services/backthink/push.go b/internal/services/backthink/push.go index 1cbf43c..ad08698 100644 --- a/internal/services/backthink/push.go +++ b/internal/services/backthink/push.go @@ -158,6 +158,11 @@ type pusher struct { gate PushGate exporter PushExporter log *slog.Logger + // audit 是可选审计出口(what/result 两字段,与 backthink 的 Auditor 同族)。 + // 它在**推送失败但渠道是主人能感知的那个**时留一行痕,因此主人不必等面板翻记录 + // 才能知道「她要发微信、微信没送到」。nil ⇒ 静默跳过:审计是留痕,不是门, + // 缺它绝不改变投递结果。 + audit func(what, result string) // exportDone 非 nil 时,会在每次异步导出完成后收到一个值。 // 它存在是为了让测试能确定性地观测该钩子;发送是非阻塞的, // 所以通道满也不会卡住导出器。 @@ -178,6 +183,13 @@ func (p *pusher) withExporter(exporter PushExporter) *pusher { return p } +// withAuditor 挂上审计出口并返回 pusher 以便链式调用。传 nil 则关闭该钩子 +// (测试里的裸 pusher 因此不产生审计)。 +func (p *pusher) withAuditor(audit func(what, result string)) *pusher { + p.audit = audit + return p +} + // Push 实现 Pusher:它套用频率门,然后把消息路由到 // 单个目标。"web" 提示到达活跃 Web/原生客户端;其它任何 // 提示("im"、"auto" 或具体平台)经 IM 桥走一次,由 @@ -205,6 +217,10 @@ func (p *pusher) Push(ctx context.Context, req PushRequest) PushResult { msg := req.Message var res PushResult + // imFault 非 nil ⇒ IM 路**确实失败过**(带着错因)。它必须活到最终结果:过去 + // 降级到活跃客户端后 res 被整条覆盖,于是「微信没送到」在那次推送里彻底消失 + // —— 正是主人反复收不到、面板却显示「已送达」的成因。 + var imFault *imUndelivered if hint == channelWeb { res = p.pushActive(ctx, msg) } else { @@ -217,10 +233,17 @@ func (p *pusher) Push(ctx context.Context, req PushRequest) PushResult { } else { msg.Platform = hint } - res = p.pushIM(ctx, msg) + im, fault := p.pushIM(ctx, msg) + res = im if !res.Pushed { + // 降级路照旧兜底(主人仍要被触达),但 IM 那次的失败如实带上: + // 降级成功 ≠ 她的话送到了她选的渠道。 res = p.pushActive(ctx, msg) + if fault != nil { + res = annotateIMUndelivered(res, *fault, im.Err) + } } + imFault = fault } // 只有真正的投递才推进冷却/按风格窗口(且只对 // 受门控的推送),因此尝试过但失败的推送不会抑制之后的 @@ -238,6 +261,11 @@ func (p *pusher) Push(ctx context.Context, req PushRequest) PushResult { res.Bypassed = true res = annotateBypass(res) } + // IM 路的失败必须**有可观测出口**:一条带原因码的结构化 WARN 已在 pushIM 落过, + // 这里再落一条主人能读懂的审计(此刻才知道这一轮最终落到哪条路)。 + if imFault != nil { + p.auditIMUndelivered(*imFault, res) + } return res } @@ -305,16 +333,138 @@ func activeDeliveryDetail(d ProactiveDelivery) string { return fmt.Sprintf("delivered to %d active client(s), %s", d.Connections, state) } -// pushIM 只投递给 IM 桥。 -func (p *pusher) pushIM(ctx context.Context, msg adapter.UnifiedMessage) PushResult { +// whatPushIMUndelivered 是「IM 路未送达」审计行的动作标签(点号小写稳定标签,与 +// backthink.self_wake.* 同族;who 由 backthink.auditEvent 统一给 "backthink")。 +const whatPushIMUndelivered = "backthink.push.im_undelivered" + +// imChannelUnknown 是渠道的显示名兜底:适配器没自报渠道、且这次请求的是泛化提示 +// (auto/im,msg.Platform 只写着一个路由哨兵)时用它。它如实表示「不知道是哪个渠道」, +// 绝不拿哨兵值冒充平台名。 +const imChannelUnknown = "unknown" + +// imUndelivered 是一次 IM 路失败的结构化事实:稳定原因码、出错渠道(可能为空 = +// 未自报)、以及底层错误。它存在的理由是 PushResult.Err 是**字符串** —— 原因码一旦 +// 落成文本就再也读不出来,而日志、记录标签与审计三处都要那个码。 +type imUndelivered struct { + channel string + category adapter.DeliveryCategory + err error +} + +// newIMUndelivered 把一次 IM 路的错误解成结构化事实。渠道取适配器**自报**值 +// (沿 %w 链找 adapter.DeliveryFailure);未自报且这次请求的是具体平台时,退到 +// 该平台(此刻 msg.Platform 就是那个渠道,不是猜的);两者都没有 ⇒ 空串 = 未知。 +func newIMUndelivered(err error, platform string) *imUndelivered { + f := &imUndelivered{ + channel: adapter.DeliveryChannelOf(err), + category: adapter.DeliveryCategoryOf(err), + err: err, + } + if f.channel == "" && platform != defaultProactivePlatform { + f.channel = platform + } + return f +} + +// channelName 是渠道的显示名;未自报时为 imChannelUnknown。 +func (f imUndelivered) channelName() string { + if f.channel == "" { + return imChannelUnknown + } + return f.channel +} + +// categoryName 是原因码的显示值;没有码时如实落到兜底码 failed(与 +// adapter.DeliveryCategoryOf 同口径,绝不编一个更精确的码)。 +func (f imUndelivered) categoryName() adapter.DeliveryCategory { + if f.category == "" { + return adapter.CategoryFailed + } + return f.category +} + +// tag 是写进 PushResult.Detail 的**稳定可解析**标签 +// `im_undelivered=<渠道>:<原因码>`。管理面板的「推送说明」直接读 Detail +// (frontend/web/src/admin/modules/Backthink.tsx),因此主人不必等一次代码改动 +// 就能看到「微信为什么没送到」。 +func (f imUndelivered) tag() string { + return fmt.Sprintf("im_undelivered=%s:%s", f.channelName(), f.categoryName()) +} + +// pushIM 只投递给 IM 桥。返回的第二个值非 nil 表示**这次 IM 投递失败**(带原因码与 +// 渠道);IM 槽未接线不是渠道故障,因此不算失败事实(降级到活跃客户端是诚实的兜底)。 +// +// 失败至少落一条**带来源与原因码**的 WARN:它同时给出 category(机器可读)、channel +// (适配器自报的真实渠道)、platform(这次请求的平台)与原文错误,因此「收不到」永远 +// 有一个可观测出口 —— 它绝不静默。 +func (p *pusher) pushIM(ctx context.Context, msg adapter.UnifiedMessage) (PushResult, *imUndelivered) { if p.im == nil { - return PushResult{Pushed: false, Route: RouteNone, Detail: "no IM target"} + return PushResult{Pushed: false, Route: RouteNone, Detail: "no IM target"}, nil } if err := p.im.SendProactive(ctx, msg); err != nil { - p.log.Warn("backthink IM push failed", "error", err) - return PushResult{Pushed: false, Route: RouteIMBridge, Err: err.Error()} + fault := newIMUndelivered(err, msg.Platform) + p.log.Warn("backthink IM push failed", + "category", string(fault.categoryName()), + "channel", fault.channelName(), + "platform", msg.Platform, + "route", RouteIMBridge, + "error", err) + return PushResult{Pushed: false, Route: RouteIMBridge, Err: err.Error()}, fault + } + return PushResult{Pushed: true, Route: RouteIMBridge, Detail: "delivered via IM bridge"}, nil +} + +// annotateIMUndelivered 把 IM 路的失败如实拼进**降级后**的结果:Detail 追加稳定标签 +// (渠道 + 原因码),因此记录与面板都明说「这个渠道没送到」。 +// +// 降级路也没成时,IM 那次的错因是「这一轮为什么没送到」的主因,落进 Err;降级路自己 +// 的错因(若有)另起一段附在后面 —— 两条都不被吞,也不互相冒充。 +func annotateIMUndelivered(res PushResult, fault imUndelivered, imErr string) PushResult { + res.Detail = appendDetail(res.Detail, fault.tag()) + if res.Pushed || imErr == "" { + return res + } + switch { + case res.Err == "": + res.Err = imErr + case !strings.Contains(res.Err, imErr): + res.Err = imErr + " | fallback: " + res.Err + } + return res +} + +// auditIMUndelivered 把一次 IM 路未送达写进审计。审计是留痕不是门:出口未接线 +// (p.audit == nil,例如测试里的裸 pusher)时静默跳过,投递结果不受影响。 +func (p *pusher) auditIMUndelivered(fault imUndelivered, final PushResult) { + if p.audit == nil { + return + } + p.audit(whatPushIMUndelivered, imUndeliveredAuditText(fault, final)) +} + +// imUndeliveredAuditText 是给主人看的中文一行:哪个渠道、什么原因码、以及这一轮最终 +// 落到了哪里。它**绝不**把「降级送到活跃客户端」说成「已送达那个渠道」。 +func imUndeliveredAuditText(fault imUndelivered, final PushResult) string { + outcome := "本轮没有在线 Web 客户端,这条消息没有送达" + if final.Pushed { + outcome = "已改由 Web 活跃客户端送达" + } + return fmt.Sprintf("主动消息未能送达 %s(原因码 %s):%v;%s", + fault.channelName(), fault.categoryName(), fault.err, outcome) +} + +// appendDetail 把 tag 追加到详情串(逗号分隔),空详情则直接成为详情;空 tag 不改变 +// 详情。它是 Detail 拼接的唯一处(channel_hint / event_bypass / im_undelivered 三个 +// 标签共用同一口径,避免三份各写一遍)。 +func appendDetail(detail, tag string) string { + switch { + case tag == "": + return detail + case detail == "": + return tag + default: + return detail + ", " + tag } - return PushResult{Pushed: true, Route: RouteIMBridge, Detail: "delivered via IM bridge"} } // normalizeChannelHint 把渠道提示转小写并去空白,把同义词映射到 @@ -347,23 +497,14 @@ func annotateHint(res PushResult, hint string) PushResult { if h == "" || h == channelAuto { return res } - tag := "channel_hint=" + h - if res.Detail == "" { - res.Detail = tag - } else { - res.Detail = res.Detail + ", " + tag - } + res.Detail = appendDetail(res.Detail, "channel_hint="+h) return res } // annotateBypass 给事件驱动的推送结果打标记,使记录/面板显示它 // 跳过了 PushGate,与因限流被拒的心跳推送区分开。 func annotateBypass(res PushResult) PushResult { - if res.Detail == "" { - res.Detail = "event_bypass" - } else { - res.Detail = res.Detail + ", event_bypass" - } + res.Detail = appendDetail(res.Detail, "event_bypass") return res } diff --git a/internal/services/backthink/push_im_undelivered_test.go b/internal/services/backthink/push_im_undelivered_test.go new file mode 100644 index 0000000..abee926 --- /dev/null +++ b/internal/services/backthink/push_im_undelivered_test.go @@ -0,0 +1,214 @@ +package backthink + +import ( + "bytes" + "context" + "log/slog" + "strings" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/weixin_ilink" +) + +// 本文件锁定「主动消息没能送到微信」这件事**被如实说清楚**(主人 2026-10-07: +// 「可以,这有必要,不然我老是收不到」)。真机日志是: +// +// level=WARN msg="backthink IM push failed" ... error="channelbridge: send proactive via +// \"weixin-ilink\": channelbridge: send bubble #1 of 8: weixin_ilink: proactive push outside 24h window" +// level=INFO msg="backthink proactive push" ... route=active_client pushed=true +// reason="delivered to 2 active client(s), channel_hint=weixin-ilink" +// +// 第二行是对的,但它让第一行彻底消失:记录与面板只说「已送达」,于是「她的话没进微信」 +// 只存在于一条没人看的自由文本 WARN 里 —— 主人反复收不到,面板却是绿的。 +// +// 三条不可退让的口径(对应主人的三条要求): +// 1. **分类化 + 不假成功**:窗口关闭必须带稳定原因码,且 IM 那条路绝不进 pushed=true; +// 2. **可见**:结构化 WARN(category/channel)+ 审计 + 思考记录的 push_detail 都带上它; +// 3. **不丢消息**:IM 路失败**不得**影响活跃客户端那条路(消息照旧送达并落库)。 + +// captureLog 返回一个把日志写进缓冲区的日志器,供断言结构化字段。 +func captureLog() (*slog.Logger, *bytes.Buffer) { + buf := &bytes.Buffer{} + return slog.New(slog.NewTextHandler(buf, nil)), buf +} + +// auditRecorder 记录 pusher 审计出口的 (what, result) 对。 +type auditRecorder struct { + rows []auditRow +} + +func (a *auditRecorder) record(what, result string) { + a.rows = append(a.rows, auditRow{who: "backthink", what: what, result: result}) +} + +// TestPusher_IMWindowClosed_FallbackKeepsFailureVisible:IM 路被 24 小时窗口拒绝、 +// 活跃客户端路成功时,**两条事实都要在结果里**:消息确实送到了 Web(不假失败), +// 但微信那条路没送到、原因是 proactive_window_closed(不假成功)。 +func TestPusher_IMWindowClosed_FallbackKeepsFailureVisible(t *testing.T) { + t.Parallel() + logger, logs := captureLog() + audit := &auditRecorder{} + active := &mockActive{n: 2, persisted: true} + im := &mockIMAdapter{err: weixin_ilink.ErrProactiveWindow} + p := newPusher(active, im, &mockGate{allow: true}, logger).withAuditor(audit.record) + + res := p.Push(context.Background(), PushRequest{ + ChannelHint: "weixin-ilink", + Message: adapter.UnifiedMessage{Content: "记得喝水哦"}, + }) + + // ③ 不丢消息:活跃客户端那条路照旧成功,且拿到的是**同一条**消息。 + if !res.Pushed || res.Route != RouteActiveClient { + t.Fatalf("res = %+v, want pushed via active_client (IM 失败不得连带丢掉另一条路)", res) + } + if active.calls != 1 || active.last.Content != "记得喝水哦" { + t.Fatalf("active calls = %d last = %q, want 1 call with the message text", active.calls, active.last.Content) + } + // ① 微信未送达这件事留在结果里(面板「推送说明」直接读 Detail)。 + if !strings.Contains(res.Detail, "im_undelivered=weixin-ilink:proactive_window_closed") { + t.Fatalf("Detail = %q, want im_undelivered=weixin-ilink:proactive_window_closed", res.Detail) + } + if !strings.Contains(res.Detail, "channel_hint=weixin-ilink") { + t.Fatalf("Detail = %q, want the requested channel hint too", res.Detail) + } + if res.Err != "" { + t.Fatalf("Err = %q, want empty:消息已由活跃客户端送达,错因在 Detail 里如实标注", res.Err) + } + + // ② 结构化日志:来源(调用方注入的 logger)+ 错误类别 + 渠道 + 原文。 + log := logs.String() + for _, want := range []string{ + `msg="backthink IM push failed"`, + "category=proactive_window_closed", + "channel=weixin-ilink", + "route=im_bridge", + "outside 24h window", + } { + if !strings.Contains(log, want) { + t.Fatalf("log = %q, want it to contain %q", log, want) + } + } + + // ② 审计:一条可读的行,说清渠道、原因码与这一轮最终落到哪。 + if len(audit.rows) != 1 { + t.Fatalf("audit rows = %+v, want exactly 1", audit.rows) + } + row := audit.rows[0] + if row.what != whatPushIMUndelivered { + t.Fatalf("audit what = %q, want %q", row.what, whatPushIMUndelivered) + } + for _, want := range []string{"weixin-ilink", "proactive_window_closed", "Web 活跃客户端"} { + if !strings.Contains(row.result, want) { + t.Fatalf("audit result = %q, want it to contain %q", row.result, want) + } + } +} + +// TestPusher_IMWindowClosed_NoFallbackNotPushed:两条路都没成时**绝不**报 pushed=true, +// 且 IM 那次的错因是主因(落进 Err),失败仍有日志与审计两个出口 —— 不静默。 +func TestPusher_IMWindowClosed_NoFallbackNotPushed(t *testing.T) { + t.Parallel() + logger, logs := captureLog() + audit := &auditRecorder{} + // 没有在线客户端、也没落库 ⇒ 降级路同样失败。 + active := &mockActive{n: 0, persisted: false} + im := &mockIMAdapter{err: weixin_ilink.ErrProactiveWindow} + p := newPusher(active, im, &mockGate{allow: true}, logger).withAuditor(audit.record) + + res := p.Push(context.Background(), PushRequest{ + ChannelHint: "weixin-ilink", + Message: adapter.UnifiedMessage{Content: "记得喝水哦"}, + }) + + if res.Pushed { + t.Fatalf("res = %+v, want Pushed=false (窗口关闭 + 无客户端 = 真的没送到)", res) + } + if res.Route == RouteIMBridge { + t.Fatalf("Route = %q, want it NOT to claim the IM route delivered", res.Route) + } + if !strings.Contains(res.Detail, "im_undelivered=weixin-ilink:proactive_window_closed") { + t.Fatalf("Detail = %q, want the undelivered tag", res.Detail) + } + if !strings.Contains(res.Err, "outside 24h window") { + t.Fatalf("Err = %q, want the IM failure as the primary error", res.Err) + } + if log := logs.String(); !strings.Contains(log, "category=proactive_window_closed") { + t.Fatalf("log = %q, want the categorized WARN", log) + } + if len(audit.rows) != 1 { + t.Fatalf("audit rows = %+v, want exactly 1", audit.rows) + } + if !strings.Contains(audit.rows[0].result, "没有送达") { + t.Fatalf("audit result = %q, want it to say the round was not delivered", audit.rows[0].result) + } +} + +// TestPusher_NoIMTargetIsNotChannelFailure:IM 槽未接线**不是**渠道故障(没有渠道 +// 可失败),因此不该被标成「某渠道未送达」,也不该产生审计噪声。 +func TestPusher_NoIMTargetIsNotChannelFailure(t *testing.T) { + t.Parallel() + audit := &auditRecorder{} + active := &mockActive{n: 1, persisted: true} + p := newPusher(active, nil, &mockGate{allow: true}, discardLog()).withAuditor(audit.record) + + res := p.Push(context.Background(), PushRequest{Message: adapter.UnifiedMessage{Content: "晚安"}}) + + if !res.Pushed || res.Route != RouteActiveClient { + t.Fatalf("res = %+v, want pushed via active_client", res) + } + if strings.Contains(res.Detail, "im_undelivered") { + t.Fatalf("Detail = %q, want no im_undelivered tag (nothing failed)", res.Detail) + } + if len(audit.rows) != 0 { + t.Fatalf("audit rows = %+v, want none", audit.rows) + } +} + +// TestThink_IMWindowClosed_RecordAndAuditCarryTheCode 是**端到端**探针:她真的决定 +// 推送到 weixin-ilink、桥那边因窗口关闭拒绝、Web 客户端接手 —— 落库的那条思考记录 +// (面板读侧)必须带着「微信未送达 + 原因码」,审计也必须有一行。 +func TestThink_IMWindowClosed_RecordAndAuditCarryTheCode(t *testing.T) { + t.Parallel() + logger, logs := captureLog() + audit := &fakeAuditor{} + b, _ := newBackthink(t, Config{}, Deps{ + Thinker: &mockThinker{text: "[我决定:push]\n[渠道:weixin-ilink]\n记得喝水哦"}, + State: &mockState{state: UserState{}}, + Active: &mockActive{n: 2, persisted: true}, + IM: &mockIMAdapter{err: weixin_ilink.ErrProactiveWindow}, + Gate: &mockGate{allow: true}, + Audit: audit, + Log: logger, + Now: func() time.Time { return dayAt() }, + }) + + b.think(context.Background(), heartbeatTrigger()) + + recs, err := b.List(context.Background(), 10) + if err != nil { + t.Fatalf("List: %v", err) + } + if len(recs) == 0 { + t.Fatal("no record persisted") + } + rec := recs[0].PushResult + if !rec.Pushed || rec.Route != RouteActiveClient { + t.Fatalf("record push = %+v, want pushed via active_client", rec) + } + if !strings.Contains(rec.Detail, "im_undelivered=weixin-ilink:proactive_window_closed") { + t.Fatalf("record Detail = %q, want the undelivered tag (面板「推送说明」读它)", rec.Detail) + } + if got := audit.countOf(whatPushIMUndelivered); got != 1 { + t.Fatalf("audit %q = %d 条, want 1(whats=%v)", whatPushIMUndelivered, got, audit.whats()) + } + if who := audit.rows[0].who; who != "backthink" { + t.Fatalf("audit who = %q, want backthink(沿用后台思考既有口径)", who) + } + // 日志带来源(services/backthink),因此这条失败在整机日志里可回溯。 + if log := logs.String(); !strings.Contains(log, "source=services/backthink") || + !strings.Contains(log, "category=proactive_window_closed") { + t.Fatalf("log = %q, want a sourced, categorized WARN", log) + } +} diff --git a/internal/services/backthink/sinks.go b/internal/services/backthink/sinks.go index 00094db..f75551a 100644 --- a/internal/services/backthink/sinks.go +++ b/internal/services/backthink/sinks.go @@ -44,8 +44,22 @@ func (b *Backthink) currentAuditor() Auditor { // rebuildPusherLocked 用当前槽位、频率门与推送出口替换 // 推送器。调用方必须持有 b.mu。 +// +// 审计出口挂的是 `b.auditEvent` 这个**读取当前出口**的回调,而不是一次性快照: +// SetAuditSink 是晚绑定的(网关比 backthink 晚组装),持快照会让稍后接上的网关 +// 永远收不到推送审计 —— 与「未接线则静默跳过」同一条诚实口径。 func (b *Backthink) rebuildPusherLocked() { - b.pusher = newPusher(b.deps.Active, b.deps.IM, b.gate, b.log).withExporter(b.deps.Exporter) + b.pusher = newPusher(b.deps.Active, b.deps.IM, b.gate, b.log). + withExporter(b.deps.Exporter). + withAuditor(b.auditEvent) +} + +// rebuildPusher 是不持锁版本的 rebuildPusherLocked,供 New 在结构体刚构造好、 +// 尚无并发访问者时使用。 +func (b *Backthink) rebuildPusher() { + b.mu.Lock() + defer b.mu.Unlock() + b.rebuildPusherLocked() } // currentPusher 在生命周期锁下返回当前推送器,因此一次槽位 diff --git a/internal/services/channelbridge/proactive_delivery_category_test.go b/internal/services/channelbridge/proactive_delivery_category_test.go new file mode 100644 index 0000000..316cec4 --- /dev/null +++ b/internal/services/channelbridge/proactive_delivery_category_test.go @@ -0,0 +1,75 @@ +package channelbridge + +import ( + "context" + "errors" + "strings" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/weixin_ilink" +) + +// 本文件锁定桥接器**再加一层包装**之后,适配器自报的失败原因码与渠道名依然读得出来。 +// +// 为什么值得单独钉:真机事故(2026-10-07 10:38)里主人看到的是 +// +// channelbridge: send proactive via "weixin-ilink": channelbridge: send bubble #1 of 8: +// weixin_ilink: proactive push outside 24h window +// +// —— 三层自由文本。若包装(%w 之外的任何改写)把自报信息吃掉,backthink 的日志、 +// 思考记录与审计就只能复述这段文本,面板也就永远说不出「微信因为哪条规则没送到」。 + +// TestSendProactive_WindowClosedKeepsCategory 用真实 24h 窗口哨兵穿过整条桥接器 +// 出口(多气泡包装 + 主动推送包装),验证码与渠道都还在。 +func TestSendProactive_WindowClosedKeepsCategory(t *testing.T) { + t.Parallel() + ad := &fakeAdapter{name: "weixin-ilink", connected: true, sendErr: weixin_ilink.ErrProactiveWindow} + owner := testOwnerProfile("weixin-ilink", "o9cq802QPKPscBenJAp-SMZPBlnQ@im.wechat") + b := newProactiveBridge(t, owner, ad) + + msg := proactiveMsg("
一
二
") + msg.Platform = "weixin-ilink" + err := b.SendProactive(context.Background(), msg) + if err == nil { + t.Fatal("SendProactive = nil, want the window-closed failure (no fake success)") + } + if !errors.Is(err, weixin_ilink.ErrProactiveWindow) { + t.Fatalf("err = %v, want it to still match ErrProactiveWindow", err) + } + // 证明这次确实走了「多气泡 + 主动推送」两层包装(否则本用例没覆盖要防的那条路)。 + if !strings.Contains(err.Error(), "send bubble #1 of 2") { + t.Fatalf("err = %v, want the multi-bubble wrapper (send bubble #1 of 2)", err) + } + if !strings.Contains(err.Error(), `send proactive via "weixin-ilink"`) { + t.Fatalf("err = %v, want the proactive wrapper naming the channel", err) + } + if got := adapter.DeliveryCategoryOf(err); got != "proactive_window_closed" { + t.Fatalf("category = %q, want proactive_window_closed", got) + } + if got := adapter.DeliveryChannelOf(err); got != "weixin-ilink" { + t.Fatalf("channel = %q, want weixin-ilink", got) + } +} + +// TestSendProactive_UnreportedFailureStaysUnknown 是反面探针:适配器**没有**自报 +// 原因码时,读取侧只能得到兜底码与「渠道未知」,绝不替它编一个更精确的码或渠道。 +func TestSendProactive_UnreportedFailureStaysUnknown(t *testing.T) { + t.Parallel() + ad := &fakeAdapter{name: "weixin-ilink", connected: true, sendErr: errors.New("boom")} + owner := testOwnerProfile("weixin-ilink", "owner-1") + b := newProactiveBridge(t, owner, ad) + + msg := proactiveMsg("只有一段") + msg.Platform = "weixin-ilink" + err := b.SendProactive(context.Background(), msg) + if err == nil { + t.Fatal("SendProactive = nil, want the failure") + } + if got := adapter.DeliveryCategoryOf(err); got != adapter.CategoryFailed { + t.Fatalf("category = %q, want the fallback %q", got, adapter.CategoryFailed) + } + if got := adapter.DeliveryChannelOf(err); got != "" { + t.Fatalf("channel = %q, want empty (unknown) — 不得替适配器猜渠道", got) + } +} diff --git a/plugins/channels/weixin_ilink/adapter.go b/plugins/channels/weixin_ilink/adapter.go index 20be2b0..83ed1e6 100644 --- a/plugins/channels/weixin_ilink/adapter.go +++ b/plugins/channels/weixin_ilink/adapter.go @@ -66,21 +66,71 @@ const ( errcodeSessionRejected = -14 ) -// 本适配器返回的哨兵错误,可用 errors.Is 判定。 +// 本适配器的投递失败原因码(adapter.DeliveryCategory,见该包 delivery.go 的口径)。 +// 码值属于**本平台**(平台规则各有各的名字),因此在这里声明、由下面的 +// selfDescribingError 自报;契约层只提供读取函数,不枚举它们。 +const ( + // categoryProactiveWindowClosed 是**平台规则**导致主动推送被拒:iLink 只允许在用户 + // 最后一条消息之后 24 小时内主动推送。它不是崩溃,也不是可以重试解决的失败。 + categoryProactiveWindowClosed = adapter.DeliveryCategory("proactive_window_closed") + // categoryNoContextToken 是这次发送没有任何可用的会话凭证(对端 context_token 缺失): + // 平台无从把消息关联到会话,因此**不能**硬发。 + categoryNoContextToken = adapter.DeliveryCategory("no_context_token") + // categorySessionRejected 是 iLink 已拒绝这份会话凭据(errcode=-14):必须重新扫码, + // 与「这次网络不通」是两类失败。 + categorySessionRejected = adapter.DeliveryCategory("session_rejected") +) + +// selfDescribingError 给哨兵错误附上**稳定原因码与渠道名**(实现 +// adapter.DeliveryFailure),使 channelbridge / backthink 能在日志、思考记录与审计里 +// 说清「哪个渠道、因为什么」没送到,而不是只能复述一整串自由文本。 +// +// Error()/Unwrap() 逐字委托被包装的哨兵 ⇒ 错误文本与 `errors.Is(err, ErrXxx)` 判定 +// 与包装之前**完全一致**(这是它能替换 errors.New 而不改变任何既有行为的原因)。 +type selfDescribingError struct { + category adapter.DeliveryCategory + err error +} + +func (e *selfDescribingError) Error() string { return e.err.Error() } +func (e *selfDescribingError) Unwrap() error { return e.err } + +// DeliveryCategory 实现 adapter.DeliveryFailure。 +func (e *selfDescribingError) DeliveryCategory() adapter.DeliveryCategory { return e.category } + +// DeliveryChannel 实现 adapter.DeliveryFailure:本适配器只说一个平台,因此渠道名恒为 +// 它自己的平台名(不是猜的,也不是调用方传进来的)。 +func (e *selfDescribingError) DeliveryChannel() string { return platformName } + +// 本适配器返回的哨兵错误,可用 errors.Is 判定(带原因码的三个由 +// selfDescribingError 包装,判定与文本不变,另附 stable 码)。 var ( // ErrInvalidConfig 在必填配置字段为空时返回。 ErrInvalidConfig = errors.New("weixin_ilink: invalid config") // ErrLoginTimeout 在二维码未及时确认时返回。 ErrLoginTimeout = errors.New("weixin_ilink: qr login timed out") - // ErrSessionRejected 在 iLink 上报 errcode=-14 时返回。 - ErrSessionRejected = errors.New("weixin_ilink: iLink rejected the session (errcode=-14); re-scan required") + // ErrSessionRejected 在 iLink 上报 errcode=-14 时返回(原因码 session_rejected: + // 主人必须重新扫码)。 + ErrSessionRejected = &selfDescribingError{ + category: categorySessionRejected, + err: errors.New("weixin_ilink: iLink rejected the session (errcode=-14); re-scan required"), + } // ErrNeedsLogin 由 Connect 在尚未配置 bot_token 时返回;扫码绑定必须经面板的 // login-qr/login-status 端点驱动。 ErrNeedsLogin = errors.New("weixin_ilink: bot not logged in yet, scan the QR") - // ErrNoContextToken 在一次发送缺少该对端的 context_token 时返回。 - ErrNoContextToken = errors.New("weixin_ilink: no context_token for peer") - // ErrProactiveWindow 在主动推送落在 24 小时窗口之外时返回。 - ErrProactiveWindow = errors.New("weixin_ilink: proactive push outside 24h window") + // ErrNoContextToken 在一次发送缺少该对端的 context_token 时返回(原因码 + // no_context_token)。 + ErrNoContextToken = &selfDescribingError{ + category: categoryNoContextToken, + err: errors.New("weixin_ilink: no context_token for peer"), + } + // ErrProactiveWindow 在主动推送落在 24 小时窗口之外时返回(原因码 + // proactive_window_closed)。它是**平台规则**,不是崩溃:调用方必须把它如实记为 + // 「未送达 + 原因码」,绝不重试刷屏、绝不经别的渠道冒充微信。 + ErrProactiveWindow = &selfDescribingError{ + category: categoryProactiveWindowClosed, + err: errors.New("weixin_ilink: proactive push outside 24h window"), + } // ErrMediaTooLarge 在下载超过 MaxMediaSize 时返回。 ErrMediaTooLarge = errors.New("weixin_ilink: media exceeds size limit") // ErrNoMediaKey 在入站附件既没有按条 aes_key、也无法用 bot_token 兜底时返回。 diff --git a/plugins/channels/weixin_ilink/delivery_failure_test.go b/plugins/channels/weixin_ilink/delivery_failure_test.go new file mode 100644 index 0000000..2d9c9d7 --- /dev/null +++ b/plugins/channels/weixin_ilink/delivery_failure_test.go @@ -0,0 +1,80 @@ +package weixin_ilink + +import ( + "context" + "errors" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// 本文件锁定「微信这次发不出去」的**稳定原因码**:24 小时窗口关闭是**平台规则**, +// 不是崩溃,因此它必须能以机器可读的码(proactive_window_closed)+ 渠道名交给上层, +// 而不是只留一整串自由文本(真机日志 2026-10-07:「主动消息没能送到微信」在面板上 +// 只表现为一句英文)。 +// +// 🔴 口径:错误文本与 errors.Is 判定**逐字不变**(自描述包装只加码,不改语义), +// 因此本文件同时是「加码没改行为」的回归探针。 + +// TestSendProactive_ExpiredWindowCarriesCategory 走**真实适配器路径**(httptest 假 +// 服务端 + 已确认连接 + 窗口外的时钟)验证原因码与渠道名,而不是只测哨兵值本身。 +func TestSendProactive_ExpiredWindowCarriesCategory(t *testing.T) { + t.Parallel() + rec := &reqRecorder{} + srv := sendOKServer(t, rec) + ad := newConnectedAdapter(t, Config{BaseURL: srv.URL, BotToken: "tok"}, srv) + + now := time.Date(2026, 9, 6, 12, 0, 0, 0, time.UTC) + ad.clock = func() time.Time { return now } + ad.recordInbound("tok-p", adapter.UnifiedMessage{SenderID: "u-1", Timestamp: now}) + now = now.Add(25 * time.Hour) + + err := ad.SendProactive(context.Background(), adapter.UnifiedMessage{ + SenderID: "u-1", Content: "too late", + }) + if !errors.Is(err, ErrProactiveWindow) { + t.Fatalf("err = %v, want ErrProactiveWindow", err) + } + if got := adapter.DeliveryCategoryOf(err); got != "proactive_window_closed" { + t.Fatalf("category = %q, want proactive_window_closed", got) + } + if got := adapter.DeliveryChannelOf(err); got != platformName { + t.Fatalf("channel = %q, want %q", got, platformName) + } + // 文本与包装之前逐字一致:加码**没有**改任何既有语义。 + if want := "weixin_ilink: proactive push outside 24h window"; err.Error() != want { + t.Fatalf("err text = %q, want %q", err.Error(), want) + } +} + +// TestSentinelErrorsCarryCategories 是三个可操作失败的原因码表:窗口关闭 +// (平台规则)、缺少会话凭证(无从关联会话)、会话被拒(必须重新扫码)。其余哨兵 +// 不自报码 —— 它们落到契约层兜底码 failed,而不是被安一个更精确的假原因。 +func TestSentinelErrorsCarryCategories(t *testing.T) { + t.Parallel() + tests := []struct { + name string + err error + want adapter.DeliveryCategory + wantChannel string + }{ + {name: "主动推送窗口关闭", err: ErrProactiveWindow, want: "proactive_window_closed", wantChannel: platformName}, + {name: "缺少会话凭证", err: ErrNoContextToken, want: "no_context_token", wantChannel: platformName}, + {name: "会话被拒(需重新扫码)", err: ErrSessionRejected, want: "session_rejected", wantChannel: platformName}, + // 未自报的哨兵连渠道也不报:读取侧得到空串(= 未知),不替它猜平台名。 + {name: "未自报的哨兵落兜底", err: ErrMediaTooLarge, want: adapter.CategoryFailed, wantChannel: ""}, + } + for _, tc := range tests { + tc := tc + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if got := adapter.DeliveryCategoryOf(tc.err); got != tc.want { + t.Fatalf("category = %q, want %q", got, tc.want) + } + if got := adapter.DeliveryChannelOf(tc.err); got != tc.wantChannel { + t.Fatalf("channel = %q, want %q", got, tc.wantChannel) + } + }) + } +}