diff --git a/Docs/api-client.md b/Docs/api-client.md index 3ce3b0d..edb3f97 100644 --- a/Docs/api-client.md +++ b/Docs/api-client.md @@ -110,7 +110,7 @@ - Web 端镜像实现在 `frontend/web/src/lib/enums.ts`（`labelFor(kind, value)` 对未知值返回原值），权威仍以本端点为准。 - **反漂移保证（后端）**：后端有测试断言「每个内置适配器的 `Name()` 都在 `platforms` 表内」「`moods` 与 persona 情绪状态机的内置状态一致」「`modes` 与前端 `MODES` 九模式一致」——漏登记即 CI FAIL，客户端拿到的表不会滞后于代码。 -> **契约版本 `v0.38`**（新增端点，向后兼容）。 +> **契约版本 `v0.39`**（新增可选字段，向后兼容）。 ## 2. REST 端点（管理，`/api/v1/admin/*`） @@ -124,7 +124,7 @@ | 4 | `/api/v1/admin/plugins` | GET | ✅ | 插件/能力列表（含 MCP/渠道） | | 5 | `/api/v1/admin/plugins/policy` | POST | ✅ | 写工具策略（owner-only，审计） | | 5a | `/api/v1/admin/plugins/installed` | GET | ✅ | **已装运行时插件列表**（owner/admin 读）：`{data:[{id,name,kind,transport,endpoint?,command?,args?,apiKey?,env?,createdAt?,live,toolCount}]}`，`apiKey`/敏感 `env` 值**掩码**（如 `****1234`），`live` = 是否已连接注册，`toolCount` = 该 server 注册工具数 | -| 6 | `/api/v1/admin/channels` | GET | ✅ | 渠道列表/健康 + 每平台已配账号（脱敏 token 掩码，design/04 §3.1）；已连接渠道的账号附 **bot 身份**（`botId`/`botUsername`/`botNickname`，如 KOOK `GET /api/v3/user/me` 自动获取） | +| 6 | `/api/v1/admin/channels` | GET | ✅ | 渠道列表/健康 + 每平台已配账号（脱敏 token 掩码，design/04 §3.1）；已连接渠道的账号附 **bot 身份**（`botId`/`botUsername`/`botNickname`，如 KOOK `GET /api/v3/user/me` 自动获取）；适配器要求重新授权时附 **`needsRescan`** + **`lastError`**（v0.39，仅微信 iLink 过期 token） | | 7 | `/api/v1/admin/iot` | GET | ✅ | IoT 设备列表 | | 8 | `/api/v1/admin/users` | GET/POST | ✅(GET) | 用户/角色/API key（GET 已实现；POST 创建用户/签发 key 随 M10 落地） | | 9 | `/api/v1/admin/llm` | GET/POST | ✅ | LLM 多厂商多模型（models.json）：GET 列表（APIKey 脱敏，含 `provider` 协议与 `maxTokens` 预算）；POST 保存组→落盘 `models.json`→**热更新**运行中的 LLM stack（design/04 §3.1，无需重启） | @@ -160,8 +160,9 @@ - `DELETE /admin/automations/{id}` → `200 {data:{id,deleted:true}}`；`404`。 - `POST /admin/llm` body `{id?,provider?,baseUrl,apiKey?,model,multimodal?,fallback?,maxTokens?}` → `201 {data:{provider:{id,name,provider,baseUrl,model,multimodal,enabled,fallback?,maxTokens?}, persisted:true, path}}`。**真实持久化**：读-改-写 `models.json`（本地、gitignore），保存组提升为**生效组**（index 0），并**热更新**运行中的 LLM provider（`path` 为落盘文件路径）；API key 写入文件但**永不回显/不落日志**。`provider` 为空默认 `openai`；缺 `baseUrl`/`model` 回 `400`；`maxTokens` 为负回 `400`；未注入 LLM stack 回 `500 llm_unavailable`、写盘失败 `500 llm_persist_failed`、热重载失败 `500 llm_reload_failed`。`maxTokens` 缺省/`0` = 未设（回落 `LLM_FALLBACK_MAX_TOKENS` / 适配器默认）。`.env` 为兜底。 - `POST /admin/llm/test` body `{provider?,baseUrl,model,apiKey?,prompt?}` → `200 {data:{ok,provider,model,reply,tokens}}`（**真实模型调用**）；无效/失败回 `400 invalid_provider` / `500 llm_test_failed`。API key 仅本次请求内使用，不落库、不进 wire。 -- `POST /admin/channels/open`、`POST /admin/channels/close` body `{name}` → `200 {data:{name,connected}}`（真实 Connect/Disconnect）；`404`（未注册渠道）/`500`（凭据缺失等连接失败）。 +- `POST /admin/channels/open`、`POST /admin/channels/close` body `{name}` → `200 {data:{name,connected}}`（真实 Connect/Disconnect）；`404`（未注册渠道）/`409 session_expired`（微信 iLink bot token 已过期，重连无法修复，须先「扫码登录」重新绑定，v0.39）/`409 needs_login`（微信 iLink 尚无 token）/`500`（其它凭据缺失等连接失败）。 - `POST /admin/channels/test` body `{platform?,accountId,isGroup}` → `200 {data:{role,owner,isGroup,env,verdict}}`。**结构性判定**：channel 层 actor 元数据 × `configs/users/owner.yaml` 白名单，绝不靠提示词。 +- **微信 iLink「会话过期」契约**（v0.39，适配器级，非端点）：iLink 返回 `errcode=-14`（bot token 过期）时适配器进入**显式过期态**——`IsConnected()` 恒 `false`；`Connect()` **不再返回成功**，而是返回「需要重新扫码」类错误（`POST /admin/channels/open` 对应 `409 session_expired`）；`HealthCheck` 同样把「过期」与「未连接」区分开。`GET /admin/channels` 该平台行带 **`needsRescan:true`** + **`lastError`**（人类可读原因，**绝不含 token**）。**清除条件**：扫码登录拿到新 token（`login-status` 报 `confirmed`）或配置里换了新 token（保存后热重载）。过期期间：`login-status` **不**假报 `confirmed` 而是 `failed` + 过期原因（避免面板把死 token 当成功重新落盘）；`login-qr` 仍可正常出码（**不**回 `already_logged_in` 挡住重新扫码）；`get_bot_qrcode` 不再携带已失效 token（否则服务端会以「已绑定」应答而不发新凭据）。未实现该能力的平台：`needsRescan`/`lastError` **不出现**（`omitempty`），客户端按缺省处理。 - **KOOK 适配器面板端点**（design/17 §2，**owner-only + 审计**，仅 `kook` 平台实现；其它平台回 `501 not_implemented`，账号未找到回 `404`）：这些端点均带可选 `?accountId=` 定位到具体 channels.json 账号（不传=该平台首个账号）。**token/密钥不打印、不落日志**；子项拉取失败不伪造、记入 `errors`。 - `GET /admin/channels/{platform}/view?accountId=` → `200 {data:{platform,accountId?,friends?,userChats?,guilds?,channels?,members?,errors?}}`。聚合该账号真实数据（好友/申请/屏蔽、服务器、首个服务器频道与成员、私信会话）；`errors` 为 `{子项:原因}` 记录失败的子项。 - `GET /admin/channels/{platform}/friends?type=&accountId=` → `200 {data:{friends?,requests?,blocked?}}`。`type` ∈ `friend|request|block`（空=全部），调用 KOOK `GET /friend?type=`。 @@ -327,6 +328,7 @@ - 语音：上行 `voice_input`（base64 音频或已转写 text），下行 `voice_output`（合成音频 base64 + text）。 ## 5. 变更记录 +- `v0.39`（**微信 iLink 会话过期可见化 + 不再假 OK**，对客户端**新增两个可选字段、一处错误码**，向后兼容）：① **`GET /api/v1/admin/channels` 每平台行新增可选 `needsRescan`（bool，`omitempty`）+ `lastError`（string，`omitempty`）**——仅当适配器**真的**要求重新授权时出现（今天只有微信 iLink 的 `errcode=-14` 过期 token 这一种），取值来自适配器新增的**可选能力接口** `RescanRequired() (bool, string)`，网关**经类型断言**读取（主 `PlatformAdapter` 契约**不变**；未实现该能力的平台字段**缺省不出现**）。用途：把「需要重新扫码」与「主人自己关掉的未连接」区分开——后者仅 `connected:false`。② **`POST /admin/channels/open`** 对「已过期」回 **`409 session_expired`**（message 明确要求用「扫码登录」重新绑定），**不再**吞掉错误回 200；`409 needs_login`（尚无 token，v0.8 起）不变。③ **适配器行为**：iLink `errcode=-14` 进入**显式过期态**——`IsConnected()` 恒 `false`；`Connect()` 返回过期错误（**不再**对已过期 token 报成功）；`HealthCheck` 区分「过期」与「未连接」；并以 **`ERROR`** 级日志写明「需要重新扫码登录」（只带 `bot_id` 等非敏感标识，**token 永不入日志**）。过期态在读循环与发送两条路径**都会**置位，并仅在**扫码登录拿到新 token** 或**配置换了新 token**（保存 + 热重载）时清除。④ 扫码登录侧：过期期间 `login-status` 报 **`failed` + 过期原因**（**不再**假报 `confirmed`，否则面板会把死 token 当成功重新落盘）；`login-qr` 仍可正常出码（**不**回 `already_logged_in`，否则重新扫码这条路被堵死）；`get_bot_qrcode` 不再携带已失效的 token（否则服务端以「已绑定」应答、**不**发新凭据）。**向后兼容**：只新增可选字段与一处更诚实的错误码；不读新字段、不处理 409 的客户端行为不变（此时面板显示「未连接」，仍可用「扫码登录」恢复）。详见 §2 行 6 与「微信 iLink『会话过期』契约」。 - `v0.38`（**B4 客户端共享枚举权威表 + 未知值回落约定**，对客户端**新增一个只读 REST 端点**）：新增 **`GET /api/v1/meta/enums`**（**任意已认证角色可读**，与 `GET /sessions` 同级；无 token `401`）→ `200 {data:{platforms,moods,modes,unknown}}`，每项 `{id,label}`：① **`platforms`**（6）= `web`/`kook`/`qq-obv11`/`weixin-ilink`/`telegram`/`webhook`——即各内置适配器 `Name()` + Web 主会话来源 `web`；② **`moods`**（5）= `calm`/`happy`/`excited`/`sad`/`angry`，与 persona 情绪状态机内置状态一致（该状态机**可扩展**，故非封闭集合）；③ **`modes`**（9）= `chat`/`code`/`home`/`work`/`docs`/`voice`/`vision`/`agents`/`settings`（design/07 §1）；④ **`unknown`** 恒 `"display-raw"` = **未知值回落约定机器可读**：客户端遇不在表中的取值**原样显示原始字符串**（**不丢弃、不报错、不占位符**），**服务端新增取值视为非破坏性变更**（forward-compatible）。**单一事实来源在 Go 侧**（`internal/contract/enums`），并有**反漂移测试**断言适配器 `Name()` / 情绪内置状态 / 前端九模式与三张表一致，漏登记即 CI FAIL。Web 侧镜像 `frontend/web/src/lib/enums.ts`（`labelFor` 未知值返回原值），顶栏/侧栏/消息气泡的「心情」标签已改用它渲染。**向后兼容**：仅新增端点，既有契约形状不变。详见 §1 行 15 与 §1.1。 - `v0.37`（**D4 未读状态 / 已读游标**，对客户端**新增一个语义字段、一个 REST 端点、一个下行事件**）：① `chat_sessions` 新增 **`read_at INTEGER NOT NULL DEFAULT 0`**（迁移 **0010**，down 可逆；**毫秒 Unix 时间戳**，与 `chat_messages.created_at` **同单位**）。② **语义**：**会话级单一已读游标 + 单 owner 账号**——同一账号所有端共用一个游标，**读一处 = 全端已读**（不做 per-device/per-user 游标，与「同一账号所有浏览器都属于同一 owner」一致）；`createdAt <= readAt` 的消息算已读，未读数 = `createdAt > readAt` 的条数。③ **`GET /api/v1/sessions` 每项新增恒返回的 `unreadCount`**（整数，0 = 无未读）：与 `lastMessage` **同一条 SQL 派生**（关联子查询 `COUNT(*) WHERE session_id = s.id AND created_at > s.read_at`，走已有 `idx_chat_msgs_session_created`，**无 N+1、不新增索引**）。④ **新增 `POST /api/v1/sessions/{id}/read`**（owner/admin，与 `PATCH /sessions/{id}` 同鉴权）：body `{before?}`（可选毫秒；缺省= 现在）→ `200 {data:{sessionId,readAt}}`；`before` 为 0/负数/非数字/body 非对象 → `400`，未知会话（含隔离 `sess-xxx`）→ `404`，未注入 HistoryStore → `501`。**游标只前进不后退**（`read_at = MAX(read_at, before)`），乱序/并发/重试请求幂等安全。⑤ **新增下行事件 `session_read`**：`{type:"session_read", sessionId, data:{readAt}}`，全连接广播 + 客户端按 `sessionId` 过滤，用于同账号其它端清红点；失败路径不广播（§3.9）。⑥ **升级回填**：0010 把既有会话游标回填为「该会话现有最新消息时间」，**升级当天不误报未读**（既有历史全部视为已读）。⑦ `HistoryStore`（网关消费侧契约）新增 `MarkSessionRead(ctx, sessionID, before)`；`SessionMeta` 新增 `UnreadCount`。**向后兼容**：仅新增字段/端点/事件，既有契约形状不变；不消费 `unreadCount`/`session_read` 的客户端行为不变。详见 §1 行 4/8a 与 §3.9。 - `v0.36`（**A1 流式帧附加会话 / 回合标识**，对客户端**新增两个可选字段、无端点变更**）：结构化流帧 `message.start/delta/end`、`thinking.start/delta/end`、`tool_call.start/delta/end`、`tool_execution.update` 现在带 `sessionId` + `turnId`（**向后兼容**：只新增帧字段，`data` 结构与既有事件形状不变）。`sessionId` 由 loop 在**回合边界**从运行上下文注入（模型适配器在 loop 之下、天然不感知会话，保持零值；loop 外无会话上下文时该字段为空）；`turnId` 由网关**每回合**新生成（`newTurnID`）并经 `loop.WithTurnID` 挂进运行上下文，`runTurn` 是编排层与 legacy 直连路径**唯一**的回合收口点，两路都覆盖——同一会话同时只有一个 in-flight 回合，故同轮内所有帧的 `turnId` 一致。此前这几帧属于 design/16 §4.2 标注的「若知道就带上」缺口（v0.10 起记录）：**多会话并发流式时前端会把别的会话的 delta 接进当前会话气泡**，现在客户端按 `sessionId === activeSessionId` 过滤即可路由到激活会话。`stream_chunk` / `stream_end` 的 `sessionId`/`turnId` 行为不变。详见 §3.1。 diff --git a/internal/services/channelbridge/inbound.go b/internal/services/channelbridge/inbound.go index ef8f209..d775209 100644 --- a/internal/services/channelbridge/inbound.go +++ b/internal/services/channelbridge/inbound.go @@ -22,11 +22,41 @@ import ( // deliver through the same pipeline as push-style adapters. Message processing // uses the bridge-lifetime context (not readCtx) so a reload leaves in-flight // messages alone. +// +// When the read path returns with readCtx still alive, the adapter stopped +// reading on its own — today that is the weixin-iLink expired session +// (errcode=-14, see plugins/channels/weixin_ilink/session_state.go), which used to +// leave the bridge claiming the channel was connected while it could no longer +// deliver anything. A cancelled readCtx is a deliberate stop (bridge Stop or a +// hot reload of this name) and is handled by the caller that cancelled it. func (b *Bridge) readLoop(a adapter.PlatformAdapter, readCtx context.Context) { defer b.wg.Done() a.ReadMessages(readCtx, func(msg adapter.UnifiedMessage) { b.onMessage(b.ctx, msg, a) }) + if readCtx.Err() == nil { + b.markReadStopped(a) + } +} + +// markReadStopped records that an adapter's read loop ended by itself while the +// adapter reports no live session, so the per-name accounting (IsConnected / +// ConnectedCount) stops claiming a channel that cannot deliver. It only ever +// clears the flag — never sets it — so it cannot make a connected channel look +// alive, and it is a no-op when the adapter still reports a live session (a +// reconnect that already happened is never undone by an older loop's exit). +func (b *Bridge) markReadStopped(a adapter.PlatformAdapter) { + if a.IsConnected() { + return + } + name := a.Name() + b.mu.Lock() + defer b.mu.Unlock() + if !b.connected[name] { + return + } + b.connected[name] = false + b.log.Warn("channel read loop ended, adapter is no longer connected", "channel", name) } // onMessage is the adapter-facing inbound callback. It runs the pipeline in a diff --git a/internal/services/channelbridge/readloop_test.go b/internal/services/channelbridge/readloop_test.go new file mode 100644 index 0000000..20f8f62 --- /dev/null +++ b/internal/services/channelbridge/readloop_test.go @@ -0,0 +1,128 @@ +package channelbridge + +import ( + "context" + "sync" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// deadSessionAdapter mimics a channel whose session dies while the bridge is +// running — today the weixin-iLink expired bot token (iLink errcode=-14): Connect +// reports success, the read loop then returns on its own (nothing cancelled it) +// and IsConnected reports false from that point on. +type deadSessionAdapter struct { + name string + mu sync.Mutex + up bool +} + +func (a *deadSessionAdapter) Name() string { return a.name } +func (a *deadSessionAdapter) Config() map[string]any { return map[string]any{"bot_token": "tok"} } +func (a *deadSessionAdapter) Connect(context.Context) error { + a.mu.Lock() + a.up = true + a.mu.Unlock() + return nil +} +func (a *deadSessionAdapter) Disconnect(context.Context) error { + a.mu.Lock() + a.up = false + a.mu.Unlock() + return nil +} +func (a *deadSessionAdapter) IsConnected() bool { + a.mu.Lock() + defer a.mu.Unlock() + return a.up +} +func (a *deadSessionAdapter) HealthCheck(context.Context) error { return adapter.ErrNotConnected } +func (a *deadSessionAdapter) ReadMessages(context.Context, func(adapter.UnifiedMessage)) { + // The session died: the read loop ends by itself, no cancellation involved. + a.mu.Lock() + a.up = false + a.mu.Unlock() +} +func (a *deadSessionAdapter) SendMessage(context.Context, adapter.UnifiedResponse) error { + return adapter.ErrNotConnected +} +func (a *deadSessionAdapter) SendProactive(context.Context, adapter.UnifiedMessage) error { + return adapter.ErrNotConnected +} +func (a *deadSessionAdapter) SetMessageHandler(adapter.MessageHandler) {} +func (a *deadSessionAdapter) Capabilities() adapter.Capabilities { + return adapter.DefaultCapabilities() +} +func (a *deadSessionAdapter) ListChannels(context.Context) ([]adapter.ChannelInfo, error) { + return nil, nil +} + +// signaledAdapter is a fakeAdapter whose read loop reports when it has returned, +// so a test can wait for a replaced adapter's exit deterministically. +type signaledAdapter struct { + *fakeAdapter + exited chan struct{} +} + +func (a *signaledAdapter) ReadMessages(ctx context.Context, _ func(adapter.UnifiedMessage)) { + <-ctx.Done() + close(a.exited) +} + +// TestReadLoopExitClearsConnectedFlag is the regression for "the adapter goes mute +// while the bridge still claims it is connected": the read loop ends with the +// session down and nothing cancelled it, so IsConnected/ConnectedCount must stop +// reporting a live channel. +func TestReadLoopExitClearsConnectedFlag(t *testing.T) { + t.Parallel() + b := newReloadBridge(t, nil) + ad := &deadSessionAdapter{name: "weixin-ilink"} + b.Add(ad) + b.Start() + + waitDeadline := time.Now().Add(2 * time.Second) + for time.Now().Before(waitDeadline) { + if !b.IsConnected("weixin-ilink") { + if n := b.ConnectedCount(); n != 0 { + t.Fatalf("ConnectedCount() = %d, want 0 after the only read loop died", n) + } + return + } + time.Sleep(time.Millisecond) + } + t.Fatal("bridge still reports weixin-ilink as connected after its read loop ended") +} + +// TestReplacedAdapterExitKeepsReplacementConnected guards the other direction: the +// read-loop exit accounting must not undo a replacement installed by a hot reload. +// The replaced adapter's loop is cancelled by Reload (a deliberate stop), so its +// exit says nothing about the adapter that now owns the name. +func TestReplacedAdapterExitKeepsReplacementConnected(t *testing.T) { + t.Parallel() + old := &signaledAdapter{ + fakeAdapter: &fakeAdapter{name: "kook:conf0", cfg: map[string]any{"bot_token": "old"}}, + exited: make(chan struct{}), + } + factory := &stubFactory{adpts: []adapter.PlatformAdapter{ + &fakeAdapter{name: "kook:conf0", cfg: map[string]any{"bot_token": "new"}}, + }} + b := newReloadBridge(t, factory) + b.Add(old) + b.Start() + + if err := b.Reload("kook"); err != nil { + t.Fatalf("Reload: %v", err) + } + <-old.exited + // The exit accounting runs immediately after the read path returns; a short + // settle window catches it without relying on a specific interleaving. + settle := time.Now().Add(50 * time.Millisecond) + for time.Now().Before(settle) { + if !b.IsConnected("kook:conf0") { + t.Fatal("the replaced adapter's read-loop exit cleared the replacement's connected flag") + } + time.Sleep(time.Millisecond) + } +} diff --git a/internal/services/gateway/admin.go b/internal/services/gateway/admin.go index b23ff6c..81dfeca 100644 --- a/internal/services/gateway/admin.go +++ b/internal/services/gateway/admin.go @@ -34,10 +34,16 @@ type PluginSummary struct { // carries that platform's configured accounts (masked, no raw tokens) so the // panel can show and edit the account config; it is omitted when s.channels is // not wired or the platform has no configured account. +// +// NeedsRescan/LastError (v0.39) report that the adapter needs the owner to +// re-authorize it (see rescanAware in admin_channels_rescan.go); both are omitted +// unless the adapter really asked for it. type ChannelSummary struct { - Name string `json:"name"` - Connected bool `json:"connected"` - Accounts []AdminChannelAccount `json:"accounts,omitempty"` + Name string `json:"name"` + Connected bool `json:"connected"` + NeedsRescan bool `json:"needsRescan,omitempty"` + LastError string `json:"lastError,omitempty"` + Accounts []AdminChannelAccount `json:"accounts,omitempty"` } // policyRequest is the POST /api/v1/admin/plugins/policy body. @@ -172,6 +178,15 @@ func (s *Server) adminListChannels(w http.ResponseWriter, r *http.Request) { continue } sum := ChannelSummary{Name: platform, Connected: ch.IsConnected()} + // Optional rescan capability (weixin-iLink expired token, v0.39): reported + // only when the adapter really asks for a re-scan, so a healthy adapter + // keeps the fields omitted rather than shipping a fabricated value. + if ra, ok := asRescanAware(cap.Value); ok { + if required, reason := ra.RescanRequired(); required { + sum.NeedsRescan = true + sum.LastError = reason + } + } if accts, ok := accountsByPlatform[platform]; ok && len(accts) > 0 { sum.Accounts = enrichAccountsWithBot(accts, botByPlatform[platform]) } diff --git a/internal/services/gateway/admin_channels_rescan.go b/internal/services/gateway/admin_channels_rescan.go new file mode 100644 index 0000000..c8a6e06 --- /dev/null +++ b/internal/services/gateway/admin_channels_rescan.go @@ -0,0 +1,62 @@ +package gateway + +import "reflect" + +// This file holds the optional "the adapter needs the owner to re-authorize it" +// capability surfaced by GET /api/v1/admin/channels (v0.39). It lives next to — +// not inside — the channel summary itself because the summary list is one of +// admin.go's larger handlers and the unwrap helper is reusable. + +// rescanAware is the optional capability a channel adapter exposes when its +// session needs owner re-authorization that a reconnect cannot provide. +// *weixin_ilink.Adapter implements it (RescanRequired): iLink answering +// errcode=-14 means the bot token is dead and the owner must scan a fresh QR. The +// core adapter.PlatformAdapter contract is deliberately NOT extended, so adapters +// that do not need the capability stay untouched. +// +// A value is only ever reported when the adapter really produced it — a platform +// without the capability keeps the fields omitted rather than shipping a guess. +type rescanAware interface { + // RescanRequired reports whether a new scan-code login is required, and why + // (reason is "" when it is not). + RescanRequired() (required bool, reason string) +} + +// asRescanAware reports whether a registered channel value exposes the rescan +// capability. The registration wraps an adapter in cmd/cyrene's namedAdapter — an +// unexported struct embedding adapter.PlatformAdapter — which promotes only the +// interface's methods, so the concrete adapter's extra capability is reachable +// only by unwrapping the embedded field. A direct adapter value (tests, +// single-account registrations) is accepted as-is. +func asRescanAware(v any) (rescanAware, bool) { + if ra, ok := v.(rescanAware); ok { + return ra, true + } + inner, ok := unwrapNamedAdapter(v) + if !ok { + return nil, false + } + ra, ok := inner.(rescanAware) + return ra, ok +} + +// unwrapNamedAdapter returns the adapter a registration wrapper embeds in its +// exported PlatformAdapter field. asKookAdmin and asWeixinLoginAdmin perform the +// same unwrap for their own capability subsets and could share this helper. +func unwrapNamedAdapter(v any) (any, bool) { + rv := reflect.ValueOf(v) + if !rv.IsValid() { + return nil, false + } + if rv.Kind() == reflect.Ptr { + rv = rv.Elem() + } + if rv.Kind() != reflect.Struct { + return nil, false + } + f := rv.FieldByName("PlatformAdapter") + if !f.IsValid() || !f.CanInterface() { + return nil, false + } + return f.Interface(), true +} diff --git a/internal/services/gateway/admin_channels_rescan_test.go b/internal/services/gateway/admin_channels_rescan_test.go new file mode 100644 index 0000000..b7a13e9 --- /dev/null +++ b/internal/services/gateway/admin_channels_rescan_test.go @@ -0,0 +1,242 @@ +package gateway + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/auth" + capreg "git.yeij.top/AskaEth/Cyrene_v2/internal/capability/registry" + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" + "git.yeij.top/AskaEth/Cyrene_v2/pkg/plugins/sdk" + "git.yeij.top/AskaEth/Cyrene_v2/plugins/channels/weixin_ilink" +) + +// rescanStubAdapter is a channel adapter that also exposes the optional rescan +// capability, exactly as *weixin_ilink.Adapter does once iLink answers +// errcode=-14 (expired bot token, re-scan required). +type rescanStubAdapter struct { + name string + connected bool + rescan bool + reason string + connectErr error +} + +func (s *rescanStubAdapter) Name() string { return s.name } +func (s *rescanStubAdapter) Config() map[string]any { return map[string]any{"bot_token": "configured"} } +func (s *rescanStubAdapter) Connect(context.Context) error { return s.connectErr } +func (s *rescanStubAdapter) Disconnect(context.Context) error { return nil } +func (s *rescanStubAdapter) IsConnected() bool { return s.connected } +func (s *rescanStubAdapter) HealthCheck(context.Context) error { return nil } +func (s *rescanStubAdapter) SendMessage(context.Context, adapter.UnifiedResponse) error { + return nil +} +func (s *rescanStubAdapter) SendProactive(context.Context, adapter.UnifiedMessage) error { return nil } +func (s *rescanStubAdapter) SetMessageHandler(adapter.MessageHandler) {} +func (s *rescanStubAdapter) Capabilities() adapter.Capabilities { return adapter.DefaultCapabilities() } +func (s *rescanStubAdapter) ListChannels(context.Context) ([]adapter.ChannelInfo, error) { + return nil, nil +} +func (s *rescanStubAdapter) ReadMessages(ctx context.Context, _ func(adapter.UnifiedMessage)) { + <-ctx.Done() +} +func (s *rescanStubAdapter) RescanRequired() (bool, string) { return s.rescan, s.reason } + +// namedRescanAdapter mirrors cmd/cyrene's namedAdapter: an unexported wrapper +// embedding adapter.PlatformAdapter, which promotes only the interface's methods. +// The gateway therefore has to unwrap it to reach the adapter's rescan capability. +type namedRescanAdapter struct{ adapter.PlatformAdapter } + +var ( + _ adapter.PlatformAdapter = (*rescanStubAdapter)(nil) + _ sdk.Channel = namedRescanAdapter{} +) + +// TestAdminChannelsReportsNeedsRescan asserts the panel can tell "needs a fresh +// scan" apart from plain "not connected": the expired adapter's row carries +// needsRescan + the reason, and a healthy adapter omits both rather than shipping +// a fabricated value. +func TestAdminChannelsReportsNeedsRescan(t *testing.T) { + t.Parallel() + const reason = "iLink getupdates 返回 errcode=-14（bot token 已过期）" + caps := capreg.New() + // Registered through the wrapper, the way the composition root registers it. + if err := caps.Register(capreg.Capability{ + Name: "weixin-ilink", + Kind: capreg.KindChannel, + Value: namedRescanAdapter{PlatformAdapter: &rescanStubAdapter{name: "weixin-ilink", rescan: true, reason: reason}}, + }); err != nil { + t.Fatalf("register weixin-ilink: %v", err) + } + if err := caps.Register(capreg.Capability{ + Name: "telegram", + Kind: capreg.KindChannel, + Value: &rescanStubAdapter{name: "telegram", connected: true}, + }); err != nil { + t.Fatalf("register telegram: %v", err) + } + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{Registry: caps}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + items, body := adminChannelsList(t, ts, tok) + if len(items) != 2 { + t.Fatalf("channels = %+v, want 2 rows", items) + } + byName := map[string]ChannelSummary{} + for _, it := range items { + byName[it.Name] = it + } + wx, ok := byName["weixin-ilink"] + if !ok { + t.Fatalf("no weixin-ilink row in %+v", items) + } + if wx.Connected { + t.Error("weixin-ilink connected = true, want false while the token is expired") + } + if !wx.NeedsRescan { + t.Error("weixin-ilink needsRescan = false, want true (expired token must ask for a re-scan)") + } + if wx.LastError != reason { + t.Errorf("weixin-ilink lastError = %q, want %q", wx.LastError, reason) + } + tg, ok := byName["telegram"] + if !ok { + t.Fatalf("no telegram row in %+v", items) + } + if !tg.Connected || tg.NeedsRescan || tg.LastError != "" { + t.Errorf("telegram = %+v, want a connected row without rescan fields", tg) + } + + // The healthy channel must OMIT the fields (omitempty), not send defaults: the + // panel distinguishes "adapter did not report" from "adapter reported false". + var raw []map[string]json.RawMessage + if err := json.Unmarshal(decodeBody(t, body)["data"], &raw); err != nil { + t.Fatalf("decode raw items: %v (body=%s)", err, body) + } + for _, item := range raw { + if string(item["name"]) != `"telegram"` { + continue + } + if _, present := item["needsRescan"]; present { + t.Error("telegram row carries needsRescan although the adapter never asked for a re-scan") + } + if _, present := item["lastError"]; present { + t.Error("telegram row carries lastError although there is none") + } + } + // The reason is informational text, never a credential. + if strings.Contains(string(body), `"bot_token"`) { + t.Errorf("channel list leaked a bot_token field: %s", body) + } +} + +// TestAdminChannelOpenExpiredSessionReturns409 asserts the open endpoint answers +// honestly for a session a reconnect cannot fix: 409 session_expired with the +// re-scan instruction, never a 200 that would make the panel show "connected". +func TestAdminChannelOpenExpiredSessionReturns409(t *testing.T) { + t.Parallel() + caps := capreg.New() + if err := caps.Register(capreg.Capability{ + Name: "weixin-ilink", + Kind: capreg.KindChannel, + Value: &rescanStubAdapter{ + name: "weixin-ilink", rescan: true, + reason: "iLink bot token 已过期", + connectErr: weixin_ilink.ErrSessionExpired, + }, + }); err != nil { + t.Fatalf("register weixin-ilink: %v", err) + } + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{Registry: caps}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + resp, body := doReq(t, http.MethodPost, ts.URL+"/api/v1/admin/channels/open", + map[string]any{"name": "weixin-ilink"}, tok) + if resp.StatusCode != http.StatusConflict { + t.Fatalf("open status = %d, want 409 (body=%s)", resp.StatusCode, body) + } + var eobj struct { + Code string `json:"code"` + Message string `json:"message"` + } + if err := json.Unmarshal(decodeBody(t, body)["error"], &eobj); err != nil { + t.Fatalf("decode error envelope: %v (body=%s)", err, body) + } + if eobj.Code != "session_expired" { + t.Fatalf("error code = %q, want session_expired", eobj.Code) + } + if !strings.Contains(eobj.Message, "扫码") { + t.Fatalf("error message = %q, want the re-scan instruction", eobj.Message) + } +} + +// TestAdminChannelsRescanFromRealIlinkAdapter closes the loop across the layers: +// a REAL *weixin_ilink.Adapter that iLink answered errcode=-14 for, registered the +// way the composition root registers it (wrapped), must surface needsRescan + +// lastError in the admin channel list. +func TestAdminChannelsRescanFromRealIlinkAdapter(t *testing.T) { + t.Parallel() + ilog := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + if r.URL.Path != "/ilink/bot/getupdates" { + w.WriteHeader(http.StatusNotFound) + return + } + _, _ = w.Write([]byte(`{"errcode":-14}`)) + })) + defer ilog.Close() + + ad, err := weixin_ilink.New(weixin_ilink.Config{ + BaseURL: ilog.URL, BotToken: "tok", PollInterval: time.Millisecond, + }, ilog.Client(), nil) + if err != nil { + t.Fatalf("new weixin adapter: %v", err) + } + if err := ad.Connect(context.Background()); err != nil { + t.Fatalf("connect before expiry: %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go ad.ReadMessages(ctx, func(adapter.UnifiedMessage) {}) + + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + if required, _ := ad.RescanRequired(); required { + break + } + time.Sleep(time.Millisecond) + } + if required, _ := ad.RescanRequired(); !required { + t.Fatal("real adapter never reached the rescan-required state") + } + + caps := capreg.New() + if err := caps.Register(capreg.Capability{ + Name: "weixin-ilink", + Kind: capreg.KindChannel, + Value: namedRescanAdapter{PlatformAdapter: ad}, + }); err != nil { + t.Fatalf("register weixin-ilink: %v", err) + } + _, ts, jwt := buildEnv(t, cfgDefault(), Deps{Registry: caps}) + tok := signAny(t, jwt, "acct", auth.RoleOwner) + + items, _ := adminChannelsList(t, ts, tok) + if len(items) != 1 { + t.Fatalf("channels = %+v, want one weixin-ilink row", items) + } + if items[0].Connected { + t.Error("connected = true although iLink rejected the token") + } + if !items[0].NeedsRescan { + t.Error("needsRescan = false, want true for an expired iLink session") + } + if items[0].LastError == "" { + t.Error("lastError is empty, want the re-scan reason") + } +} diff --git a/internal/services/gateway/admin_write_llm_channels.go b/internal/services/gateway/admin_write_llm_channels.go index da12e3f..4c8419e 100644 --- a/internal/services/gateway/admin_write_llm_channels.go +++ b/internal/services/gateway/admin_write_llm_channels.go @@ -236,6 +236,15 @@ func (s *Server) adminChannelOpen(w http.ResponseWriter, r *http.Request) { } if err := ch.Connect(r.Context()); err != nil { s.auditClaims(r, "channel.open", "error") + // A state a plain reconnect cannot fix must not read as a generic failure: + // the weixin-iLink bot token was rejected by iLink (errcode=-14), so the + // owner needs a fresh scan-code login. Answer 409 with that instruction + // instead of pretending the channel opened (the panel marks it connected). + if errors.Is(err, weixin_ilink.ErrSessionExpired) { + writeError(w, apperr.Wrap(apperr.ErrConflict, "session_expired", + "该渠道会话已过期（需要重新扫码登录）：请用「扫码登录」重新绑定后再连接")) + return + } // weixin-iLink with no token yet must not hang nor read as a credential // error: the panel routes the owner to the out-of-band 扫码登录 flow. if errors.Is(err, weixin_ilink.ErrNeedsLogin) { diff --git a/plugins/channels/weixin_ilink/adapter.go b/plugins/channels/weixin_ilink/adapter.go index 23a4a28..b817b2d 100644 --- a/plugins/channels/weixin_ilink/adapter.go +++ b/plugins/channels/weixin_ilink/adapter.go @@ -159,7 +159,15 @@ type Adapter struct { mu sync.RWMutex connected bool - qrcode string + // expired marks a session iLink rejected with errcode=-14: the bot token is + // dead and only a fresh scan-code login re-binds it (session_state.go). + // expiredToken records WHICH token was rejected, so a subsequently configured + // token clears the state; rescanDetail is the panel-visible reason. All three + // are guarded by mu. + expired bool + expiredToken string + rescanDetail string + qrcode string // pollBaseURL optionally overrides cfg.BaseURL for get_qrcode_status polling. // It is set when the server returns scaned_but_redirect with a redirect_host, // so subsequent status polls follow the IDC host. Guarded by mu. @@ -230,9 +238,19 @@ func (a *Adapter) Config() map[string]any { } // Connect establishes the iLink session. It is idempotent. When a bot_token is -// already configured it is treated as a live session; otherwise it runs the -// scan-code login flow (see login.go). +// already configured and usable it is treated as a live session; otherwise it +// runs the scan-code login flow (see login.go). +// +// A token iLink already rejected with errcode=-14 is NOT usable: Connect refuses +// it with ErrSessionExpired (re-scan required) instead of reporting a session +// that the very next poll would tear down, and the scan-code login flow is the +// way back (see login_admin.go StartLogin). func (a *Adapter) Connect(ctx context.Context) error { + if expired, reason := a.expiredState(); expired { + a.log.Error("ilink connect refused: session expired, 需要重新扫码登录", + "bot_id", a.botID(), "reason", reason) + return fmt.Errorf("%w: %s", ErrSessionExpired, reason) + } a.mu.RLock() connected := a.connected token := a.cfg.BotToken @@ -250,10 +268,18 @@ func (a *Adapter) Connect(ctx context.Context) error { a.mu.Lock() a.connected = true a.mu.Unlock() - a.log.Info("ilink session established", "bot_id", a.cfg.BotID) + a.log.Info("ilink session established", "bot_id", a.botID()) return nil } +// botID returns the configured bot identifier; it is the only account identifier +// safe to log (the bot token is a credential and never reaches a log line). +func (a *Adapter) botID() string { + a.mu.RLock() + defer a.mu.RUnlock() + return a.cfg.BotID +} + // Disconnect tears down the session. It is idempotent. func (a *Adapter) Disconnect(ctx context.Context) error { a.mu.Lock() @@ -274,9 +300,15 @@ func (a *Adapter) IsConnected() bool { } // HealthCheck verifies the session is usable: it must be connected and carry a -// bot_token. It does not probe the network so callers can use it cheaply after -// a reconnect without incurring a request. +// bot_token. An expired session (iLink rejected the token with errcode=-14) is +// reported as ErrSessionExpired — "re-scan required" — so a caller can tell a +// dead credential apart from a plain disconnected adapter. It does not probe the +// network so callers can use it cheaply after a reconnect without incurring a +// request. func (a *Adapter) HealthCheck(ctx context.Context) error { + if expired, reason := a.expiredState(); expired { + return fmt.Errorf("%w: %s", ErrSessionExpired, reason) + } a.mu.RLock() defer a.mu.RUnlock() if !a.connected || a.cfg.BotToken == "" { diff --git a/plugins/channels/weixin_ilink/login.go b/plugins/channels/weixin_ilink/login.go index a25a0cd..087c164 100644 --- a/plugins/channels/weixin_ilink/login.go +++ b/plugins/channels/weixin_ilink/login.go @@ -144,7 +144,9 @@ func (a *Adapter) runLogin(sess *loginSession, sessCtx context.Context) { // completeLogin persists the confirmed credentials and marks the session live. // It writes cfg under a.mu, then clears the session under loginMu (never nesting -// the two locks), and finally signals the waiting Connect path. +// the two locks), and finally signals the waiting Connect path. A confirmed scan +// delivers a fresh bot_token, which is the canonical way OUT of the expired state +// (session_state.go): the rescan flag is cleared here. func (a *Adapter) completeLogin(sess *loginSession, status qrcodeStatusResp) { a.mu.Lock() a.cfg.BotToken = status.BotToken @@ -153,6 +155,7 @@ func (a *Adapter) completeLogin(sess *loginSession, status qrcodeStatusResp) { if status.BaseURL != "" { a.cfg.BaseURL = status.BaseURL } + a.clearRescanLocked() a.connected = true a.loginStatus = LoginStatusConfirmed a.loginDetail = "" @@ -171,11 +174,15 @@ func (a *Adapter) completeLogin(sess *loginSession, status qrcodeStatusResp) { // completeBindedRedirect treats binded_redirect as an already-bound outcome: the // scanned bot is already linked to this instance, so the server issues no new -// credentials and any existing token remains valid. +// credentials and any existing token remains valid. Because NO new token arrives, +// an expired session stays expired here — binded_redirect carries no proof that a +// token iLink already rejected with errcode=-14 works again, so the adapter does +// not claim a live session on the strength of it. A confirmed scan (completeLogin) +// is what clears the expired state. func (a *Adapter) completeBindedRedirect(sess *loginSession) { a.log.Info("weixin_ilink: login already bound (binded_redirect)") a.mu.Lock() - a.connected = a.cfg.BotToken != "" + a.connected = a.cfg.BotToken != "" && !a.expiredLocked() a.loginStatus = LoginStatusConfirmed a.loginDetail = loginBindedRedirectDetail a.mu.Unlock() @@ -211,8 +218,14 @@ func (a *Adapter) failLogin(sess *loginSession, err error) { // localTokenList returns the locally-held bot tokens to advertise on // get_bot_qrcode, so an already-bound bot can be detected (binded_redirect). On a // fresh bind it returns an empty (non-nil) slice so the wire carries `[]` rather -// than `null`, matching the official client. +// than `null`, matching the official client. An EXPIRED token is deliberately not +// advertised: it is not a usable bind, and offering it lets the server answer +// binded_redirect (no new credentials) — which would leave the owner stuck with a +// dead token and no way back. Reporting `[]` asks for a real rebind. func (a *Adapter) localTokenList() []string { + if expired, _ := a.expiredState(); expired { + return []string{} + } a.mu.RLock() defer a.mu.RUnlock() if a.cfg.BotToken == "" { diff --git a/plugins/channels/weixin_ilink/login_admin.go b/plugins/channels/weixin_ilink/login_admin.go index 47a39be..92513ae 100644 --- a/plugins/channels/weixin_ilink/login_admin.go +++ b/plugins/channels/weixin_ilink/login_admin.go @@ -119,8 +119,11 @@ func (a *Adapter) StartLogin(ctx context.Context) (LoginQR, error) { a.loginMu.Lock() defer a.loginMu.Unlock() - // Already bound → scanning is moot. - if a.hasToken() { + // Already bound → scanning is moot — unless the stored token is EXPIRED: iLink + // rejected it with errcode=-14, so it is not a usable bind and the owner must + // be able to re-scan (otherwise the only way out of the expired state would be + // unreachable). A live token still short-circuits the QR flow as before. + if expired, _ := a.expiredState(); a.hasToken() && !expired { return LoginQR{}, ErrAlreadyLoggedIn } if sess := a.loginSession; sess != nil { @@ -178,11 +181,23 @@ func (a *Adapter) StartLogin(ctx context.Context) (LoginQR, error) { // LoginStatus reports the current scan-login state. A non-empty bot token // overrides to "confirmed" even if a stale session is still recorded, so the panel -// always sees the true binding. detail provides the bot id on confirmation and the -// failure reason on failure. +// always sees the true binding — except when that token is EXPIRED: iLink rejected +// it with errcode=-14, so reporting "confirmed" would make the panel re-persist a +// dead credential and call the bind a success. An expired bind is reported as +// failed with the re-scan reason instead, while a re-scan already in progress +// keeps its own state. detail provides the bot id on confirmation and the failure +// reason on failure. func (a *Adapter) LoginStatus(ctx context.Context) (status, detail string) { - a.mu.RLock() - defer a.mu.RUnlock() + a.mu.Lock() + defer a.mu.Unlock() + if a.expiredLocked() { + if a.loginStatus == LoginStatusWaiting || a.loginStatus == LoginStatusNeedVerifyCode { + // A re-scan the owner already started: report its progress, not the + // expired state it is meant to replace. + return a.loginStatus, a.loginDetail + } + return LoginStatusFailed, a.rescanDetail + } if a.cfg.BotToken != "" { return LoginStatusConfirmed, a.cfg.BotID } diff --git a/plugins/channels/weixin_ilink/poll.go b/plugins/channels/weixin_ilink/poll.go index 0ff70c4..c37e82b 100644 --- a/plugins/channels/weixin_ilink/poll.go +++ b/plugins/channels/weixin_ilink/poll.go @@ -95,10 +95,12 @@ func (a *Adapter) ReadMessages(ctx context.Context, emit func(adapter.UnifiedMes failures = 0 if resp.Errcode == errSessionExpired { - a.log.Warn("session expired (errcode=-14); stop polling, re-scan required") - a.mu.Lock() - a.connected = false - a.mu.Unlock() + // The bot token is dead: retrying cannot help and the poll loop would go + // mute for good. Record the explicit expired state (which also drops the + // connected flag and logs at ERROR) and stop the loop; the owner has to + // re-scan, and everything downstream sees ErrSessionExpired until a fresh + // token arrives. + a.markExpired(rescanReasonGetUpdates) return } if resp.Ret != 0 { diff --git a/plugins/channels/weixin_ilink/send.go b/plugins/channels/weixin_ilink/send.go index 5648025..636a3df 100644 --- a/plugins/channels/weixin_ilink/send.go +++ b/plugins/channels/weixin_ilink/send.go @@ -30,8 +30,8 @@ type sendResult struct { // context_token, so the platform can associate the reply with the right // conversation (design/08 §8). func (a *Adapter) SendMessage(ctx context.Context, resp adapter.UnifiedResponse) error { - if !a.IsConnected() { - return adapter.ErrNotConnected + if err := a.sendGuard(); err != nil { + return err } for _, m := range resp.Messages { peer, _ := recipient(m) @@ -51,8 +51,8 @@ func (a *Adapter) SendMessage(ctx context.Context, resp adapter.UnifiedResponse) // messaged, the push is rejected with ErrProactiveWindow and Capabilities. // Proactive is only true so callers know the capability exists. func (a *Adapter) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) error { - if !a.IsConnected() { - return adapter.ErrNotConnected + if err := a.sendGuard(); err != nil { + return err } peer, _ := recipient(msg) if peer == "" { @@ -73,6 +73,20 @@ func (a *Adapter) SendProactive(ctx context.Context, msg adapter.UnifiedMessage) return a.sendToOne(ctx, peer, msg) } +// sendGuard reports the error a send must surface before touching the network: an +// expired session (iLink already refused this token with errcode=-14) reads as +// ErrSessionExpired — the owner must re-scan — while any other down session stays +// ErrNotConnected. +func (a *Adapter) sendGuard() error { + if expired, reason := a.expiredState(); expired { + return fmt.Errorf("weixin_ilink: send: %w: %s", ErrSessionExpired, reason) + } + if !a.IsConnected() { + return adapter.ErrNotConnected + } + return nil +} + // sendToOne sends one normalized message to a peer as a native iLink message, // composing its item_list from any text plus uploaded media. func (a *Adapter) sendToOne(ctx context.Context, peer string, m adapter.UnifiedMessage) error { @@ -152,6 +166,10 @@ func (a *Adapter) sendNative(ctx context.Context, peer, ctxToken string, items [ return fmt.Errorf("weixin_ilink: sendmessage: %w", err) } if out.Errcode == errSessionExpired { + // The send path sees the same dead token as the poll path: record the + // expired state so the adapter stops claiming a live session and the panel + // can ask for a re-scan. + a.markExpired(rescanReasonSendMessage) return ErrSessionExpired } if out.Ret != 0 { @@ -172,6 +190,12 @@ func (a *Adapter) getUploadURL(ctx context.Context) (string, error) { bytes.NewReader(body), &raw); err != nil { return "", fmt.Errorf("weixin_ilink: getuploadurl: %w", err) } + if intOf(raw["errcode"]) == errSessionExpired { + // Media upload is part of the send path, so the same dead token is recorded + // here too rather than degrading into "no url returned". + a.markExpired(rescanReasonUploadURL) + return "", ErrSessionExpired + } u := firstString(raw, "upload_url", "url", "uploadUrl", "cdn_url") if u == "" { return "", fmt.Errorf("weixin_ilink: getuploadurl returned no url") diff --git a/plugins/channels/weixin_ilink/session_state.go b/plugins/channels/weixin_ilink/session_state.go new file mode 100644 index 0000000..bb166bf --- /dev/null +++ b/plugins/channels/weixin_ilink/session_state.go @@ -0,0 +1,92 @@ +package weixin_ilink + +// This file holds the adapter's session-expiry state machine (design/08 §4-§5). +// +// iLink answers errcode=-14 when the bot token is no longer valid. That is not a +// transient failure the read loop can retry away: the token is dead until the +// owner scans a fresh QR code. The adapter therefore keeps an EXPLICIT expired +// flag instead of silently dropping the connection, so: +// +// - the read loop cannot exit and leave the adapter quietly mute while still +// reporting "connected" (the "过期后静默变哑" defect); +// - Connect/HealthCheck/send surface ErrSessionExpired — "re-scan required" — +// rather than a fake OK or a generic "not connected"; +// - the admin panel can read RescanRequired() and show 需要重新扫码 (see +// internal/services/gateway/admin.go). +// +// Lifecycle (set once, cleared on a usable credential): +// +// SET markExpired iLink returned errcode=-14 on getupdates, +// sendmessage or getuploadurl. +// CLEAR clearRescanLocked a scan-code login delivered a new bot_token +// (completeLogin), or +// expiredLocked saw a bot_token different from the one +// iLink rejected — a config change / hot reload that +// supplied a fresh token. + +// Rescan reasons recorded on the expired state. They are short, human-readable +// and carry no credential: the bot token itself never leaves cfg. +const ( + rescanReasonGetUpdates = "iLink getupdates 返回 errcode=-14（bot token 已过期）" + rescanReasonSendMessage = "iLink sendmessage 返回 errcode=-14（bot token 已过期）" + rescanReasonUploadURL = "iLink getuploadurl 返回 errcode=-14（bot token 已过期）" +) + +// markExpired records that iLink rejected the current bot token (errcode=-14). +// It drops the connected flag so IsConnected can never keep claiming a live +// session for a dead token, and logs the transition at ERROR level exactly once +// (a failing poll loop or a repeated send must not flood the log). Only the +// non-secret bot_id and the reason are logged. +func (a *Adapter) markExpired(reason string) { + a.mu.Lock() + first := !a.expired + a.expired = true + a.expiredToken = a.cfg.BotToken + a.rescanDetail = reason + a.connected = false + botID := a.cfg.BotID + a.mu.Unlock() + + if first { + a.log.Error("ilink session expired: 需要重新扫码登录（请用「扫码登录」重新绑定）", + "bot_id", botID, "errcode", errSessionExpired, "reason", reason) + } +} + +// clearRescanLocked drops the expired state. Callers must hold a.mu. +func (a *Adapter) clearRescanLocked() { + a.expired = false + a.expiredToken = "" + a.rescanDetail = "" +} + +// expiredLocked reports whether the session is expired. It first clears the state +// when the configured bot_token is no longer the one iLink rejected, so a fresh +// token (scan login, operator config change, hot reload) restores the session +// without extra bookkeeping. Callers must hold a.mu. +func (a *Adapter) expiredLocked() bool { + if a.expired && a.expiredToken != a.cfg.BotToken { + a.clearRescanLocked() + } + return a.expired +} + +// expiredState reports the expired flag together with the rescan reason in one +// lock acquisition. reason is "" when the session is not expired. +func (a *Adapter) expiredState() (bool, string) { + a.mu.Lock() + defer a.mu.Unlock() + if !a.expiredLocked() { + return false, "" + } + return true, a.rescanDetail +} + +// RescanRequired implements the optional capability the admin panel reads by type +// assertion (internal/services/gateway/admin.go): it reports whether this adapter +// needs a new scan-code login instead of a plain reconnect, plus the +// human-readable reason (never a secret). A healthy adapter reports false with an +// empty reason. +func (a *Adapter) RescanRequired() (bool, string) { + return a.expiredState() +} diff --git a/plugins/channels/weixin_ilink/session_state_test.go b/plugins/channels/weixin_ilink/session_state_test.go new file mode 100644 index 0000000..7dd023e --- /dev/null +++ b/plugins/channels/weixin_ilink/session_state_test.go @@ -0,0 +1,279 @@ +package weixin_ilink + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "testing" + "time" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/channel/adapter" +) + +// These tests cover the explicit session-expiry state machine (session_state.go): +// iLink answering errcode=-14 must be visible as "re-scan required" instead of a +// silently mute adapter or a connect that reports success with a dead token. + +// expiringServer answers every getupdates long poll with the expired-session +// errcode, the way iLink reports a dead bot token. +func expiringServer(t *testing.T) *httptest.Server { + t.Helper() + return newJSONServer(t, nil, func(req recordedReq) handlerResult { + if req.path != "/ilink/bot/getupdates" { + return handlerResult{status: 404} + } + return handlerResult{body: jsonBody(map[string]any{"errcode": errSessionExpired})} + }) +} + +// expireByPoll drives the adapter into the expired state through the long poll: +// ReadMessages must return on errcode=-14 and the adapter must report the +// rescan-required state. +func expireByPoll(t *testing.T, ad *Adapter) { + t.Helper() + done := make(chan struct{}) + go func() { + defer close(done) + ad.ReadMessages(context.Background(), func(adapter.UnifiedMessage) {}) + }() + waitDone(t, done) + if required, reason := ad.RescanRequired(); !required || reason == "" { + t.Fatalf("RescanRequired() = (%v,%q), want (true, non-empty) after errcode=-14", required, reason) + } +} + +// newExpiredByPollAdapter builds a connected adapter and expires it via the poll +// path. +func newExpiredByPollAdapter(t *testing.T, srv *httptest.Server) *Adapter { + t.Helper() + ad := newConnectedAdapter(t, Config{BaseURL: srv.URL, PollInterval: time.Millisecond, BotToken: "tok-old"}, srv) + expireByPoll(t, ad) + return ad +} + +// assertExpiredSession asserts the observable contract of an expired session: not +// connected, rescan required with a reason, HealthCheck distinguishing expired from +// not-connected, and Connect refusing with ErrSessionExpired instead of a fake OK. +func assertExpiredSession(t *testing.T, ad *Adapter) { + t.Helper() + if ad.IsConnected() { + t.Fatal("IsConnected() = true while the session is expired") + } + required, reason := ad.RescanRequired() + if !required { + t.Fatal("RescanRequired() = false, want true") + } + if reason == "" { + t.Fatal("RescanRequired() reason is empty, want the re-scan prompt") + } + + herr := ad.HealthCheck(context.Background()) + if !errors.Is(herr, ErrSessionExpired) { + t.Fatalf("HealthCheck() = %v, want ErrSessionExpired", herr) + } + if errors.Is(herr, adapter.ErrNotConnected) { + t.Fatal("HealthCheck() also matches ErrNotConnected, want expired to be distinguishable") + } + + cerr := ad.Connect(context.Background()) + if cerr == nil { + t.Fatal("Connect() = nil on an expired session, want ErrSessionExpired (no fake OK)") + } + if !errors.Is(cerr, ErrSessionExpired) { + t.Fatalf("Connect() = %v, want ErrSessionExpired", cerr) + } + if errors.Is(cerr, ErrNeedsLogin) { + t.Fatal("Connect() = ErrNeedsLogin, want ErrSessionExpired (expired is not 'unconfigured')") + } + if ad.IsConnected() { + t.Fatal("IsConnected() = true after a refused Connect on an expired session") + } +} + +func TestSessionExpired_PollMarksRescanAndRefusesConnect(t *testing.T) { + t.Parallel() + srv := expiringServer(t) + ad := newExpiredByPollAdapter(t, srv) + assertExpiredSession(t, ad) +} + +func TestSessionExpired_SendMarksRescan(t *testing.T) { + t.Parallel() + srv := newJSONServer(t, nil, func(req recordedReq) handlerResult { + if req.method == http.MethodPost && req.path == "/ilink/bot/sendmessage" { + return handlerResult{body: jsonBody(map[string]any{"errcode": errSessionExpired})} + } + return handlerResult{status: 404} + }) + ad := newConnectedAdapter(t, Config{BaseURL: srv.URL, BotToken: "tok"}, srv) + ad.recordInbound("tok-1", adapter.UnifiedMessage{SenderID: "u-1", Timestamp: ad.clock()}) + + err := ad.SendMessage(context.Background(), adapter.UnifiedResponse{ + Messages: []adapter.UnifiedMessage{{SenderID: "u-1", Content: "hi"}}, + }) + if !errors.Is(err, ErrSessionExpired) { + t.Fatalf("SendMessage() = %v, want ErrSessionExpired", err) + } + if ad.IsConnected() { + t.Fatal("IsConnected() = true after a send saw errcode=-14") + } + // A later send reports the expired session itself, not a bare "not connected". + err = ad.SendMessage(context.Background(), adapter.UnifiedResponse{ + Messages: []adapter.UnifiedMessage{{SenderID: "u-1", Content: "again"}}, + }) + if !errors.Is(err, ErrSessionExpired) { + t.Fatalf("second SendMessage() = %v, want ErrSessionExpired", err) + } + perr := ad.SendProactive(context.Background(), adapter.UnifiedMessage{SenderID: "u-1", Content: "push"}) + if !errors.Is(perr, ErrSessionExpired) { + t.Fatalf("SendProactive() = %v, want ErrSessionExpired", perr) + } + assertExpiredSession(t, ad) +} + +func TestSessionExpired_ClearedByScanLogin(t *testing.T) { + t.Parallel() + rec := &reqRecorder{} + srv := newJSONServer(t, rec, func(req recordedReq) handlerResult { + switch req.path { + case "/ilink/bot/getupdates": + return handlerResult{body: jsonBody(map[string]any{"errcode": errSessionExpired})} + case "/ilink/bot/get_bot_qrcode": + return handlerResult{body: jsonBody(map[string]any{ + "qrcode": "qr-re", "qrcode_img_content": "https://img/qr", + })} + case "/ilink/bot/get_qrcode_status": + return handlerResult{body: jsonBody(map[string]any{ + "status": "confirmed", "bot_token": "tok-new", + "ilink_bot_id": "bot-new", "ilink_user_id": "o-1", + })} + } + return handlerResult{status: 404} + }) + ad := newExpiredByPollAdapter(t, srv) + assertExpiredSession(t, ad) + + // An expired token must not short-circuit the scan flow: it is not a usable + // bind, and ErrAlreadyLoggedIn would leave the owner no way back. + qr, err := ad.StartLogin(context.Background()) + if err != nil { + t.Fatalf("StartLogin() on an expired session = %v, want a fresh QR", err) + } + if qr.Qrcode != "qr-re" { + t.Fatalf("StartLogin() qrcode = %q, want qr-re", qr.Qrcode) + } + + // The confirmed scan delivers a new token: the expired state must clear. + waitFor(t, func() bool { ok, _ := ad.RescanRequired(); return !ok }) + if err := ad.Connect(context.Background()); err != nil { + t.Fatalf("Connect() after re-scan = %v, want nil", err) + } + if !ad.IsConnected() { + t.Fatal("IsConnected() = false after a confirmed re-scan") + } + if err := ad.HealthCheck(context.Background()); err != nil { + t.Fatalf("HealthCheck() after re-scan = %v, want nil", err) + } + if got := ad.Config()["bot_token"]; got != "tok-new" { + t.Fatalf("bot_token = %v, want tok-new", got) + } + if status, _ := ad.LoginStatus(context.Background()); status != LoginStatusConfirmed { + t.Fatalf("LoginStatus() = %q, want confirmed after a successful re-scan", status) + } + + // The re-scan asked for a real rebind: the dead token is not advertised as a + // local token (that is what invites binded_redirect with no new credentials). + reqs := rec.find(func(q recordedReq) bool { return q.path == "/ilink/bot/get_bot_qrcode" }) + if len(reqs) != 1 { + t.Fatalf("get_bot_qrcode requests = %d, want 1", len(reqs)) + } + if ltl := reqs[0].bodyJSON()["local_token_list"]; ltl == nil { + t.Fatalf("get_bot_qrcode body has no local_token_list: %s", reqs[0].body) + } else if list, ok := ltl.([]any); !ok || len(list) != 0 { + t.Fatalf("get_bot_qrcode local_token_list = %v, want an empty array", ltl) + } +} + +func TestSessionExpired_ClearedByNewToken(t *testing.T) { + t.Parallel() + srv := expiringServer(t) + ad := newExpiredByPollAdapter(t, srv) + + // The rejected token is still configured: the session stays expired. + if err := ad.Connect(context.Background()); !errors.Is(err, ErrSessionExpired) { + t.Fatalf("Connect() = %v, want ErrSessionExpired while the rejected token is unchanged", err) + } + if err := ad.HealthCheck(context.Background()); !errors.Is(err, ErrSessionExpired) { + t.Fatalf("HealthCheck() = %v, want ErrSessionExpired while the rejected token is unchanged", err) + } + + setBotToken(ad, "tok-fresh") + if required, _ := ad.RescanRequired(); required { + t.Fatal("RescanRequired() = true after a new token was configured") + } + if err := ad.Connect(context.Background()); err != nil { + t.Fatalf("Connect() with a fresh token = %v, want nil", err) + } + if !ad.IsConnected() { + t.Fatal("IsConnected() = false after Connect with a fresh token") + } + if err := ad.HealthCheck(context.Background()); err != nil { + t.Fatalf("HealthCheck() with a fresh token = %v, want nil", err) + } +} + +func TestSessionExpired_LoginStatusReportsRescanAndAllowsRescan(t *testing.T) { + t.Parallel() + srv := newJSONServer(t, nil, func(req recordedReq) handlerResult { + switch req.path { + case "/ilink/bot/getupdates": + return handlerResult{body: jsonBody(map[string]any{"errcode": errSessionExpired})} + case "/ilink/bot/get_bot_qrcode": + return handlerResult{body: jsonBody(map[string]any{ + "qrcode": "qr-2", "qrcode_img_content": "https://img/qr2", + })} + } + return handlerResult{body: jsonBody(map[string]any{"status": "wait"})} + }) + ad := newConnectedAdapter(t, Config{ + BaseURL: srv.URL, PollInterval: time.Millisecond, + LoginInterval: time.Millisecond, LoginTimeout: 30 * time.Millisecond, + BotToken: "tok-old", + }, srv) + expireByPoll(t, ad) + + // A dead token must not read as "confirmed": the panel would re-persist it and + // call the bind a success. + status, detail := ad.LoginStatus(context.Background()) + if status != LoginStatusFailed { + t.Fatalf("LoginStatus() = %q, want failed while the stored token is expired", status) + } + if detail == "" { + t.Fatal("LoginStatus() detail is empty, want the re-scan reason") + } + + // The panel's 扫码登录 must still mint a QR (ErrAlreadyLoggedIn would block the + // only recovery path). + qr, err := ad.StartLogin(context.Background()) + if err != nil { + t.Fatalf("StartLogin() = %v, want a fresh QR", err) + } + if qr.Qrcode != "qr-2" { + t.Fatalf("StartLogin() qrcode = %q, want qr-2", qr.Qrcode) + } + // While the re-scan is in progress its own state is reported, not the expired one. + if status, _ := ad.LoginStatus(context.Background()); status != LoginStatusWaiting { + t.Fatalf("LoginStatus() during the re-scan = %q, want waiting", status) + } +} + +// setBotToken swaps the configured bot token, standing in for a panel save + +// hot reload (which rebuilds the adapter) or any other in-place config refresh; +// the adapter has no public setter because production always constructs a fresh +// instance from the new config. +func setBotToken(ad *Adapter, token string) { + ad.mu.Lock() + ad.cfg.BotToken = token + ad.mu.Unlock() +}