From 529f865aba5dc78a2120f53fe636a3c9f27c2204 Mon Sep 17 00:00:00 2001 From: hit_aran Date: Wed, 5 Aug 2026 14:50:03 +0800 Subject: [PATCH 1/7] docs: design feishu observed contact enrichment --- ...eishu-observed-contact-label-enrichment.md | 53 +++++++++++++++++++ 1 file changed, 53 insertions(+) create mode 100644 docs/design/feishu-observed-contact-label-enrichment.md diff --git a/docs/design/feishu-observed-contact-label-enrichment.md b/docs/design/feishu-observed-contact-label-enrichment.md new file mode 100644 index 00000000000..60b226c757a --- /dev/null +++ b/docs/design/feishu-observed-contact-label-enrichment.md @@ -0,0 +1,53 @@ +# Feishu observed-contact label enrichment + +## Goal + +Populate Feishu observed-contact labels with the sender name and group name +without delaying inbound message processing. Keep the current ID labels when +lookup is unavailable. + +## Design + +`ChannelBase` continues to persist the ID-based observation immediately after +inbound preflight succeeds. It then invokes a synchronous, protected +post-observation hook. The default hook does nothing and its return value is +not awaited, so other channel implementations are unchanged. + +`FeishuChannel` overrides the hook and starts background lookups for IDs that +have not been attempted during the current channel instance lifetime: + +- `POST /open-apis/contact/v3/users/basic_batch` resolves the sender name. +- `GET /open-apis/im/v1/chats/:chat_id` resolves the group name. + +User and group lookups have separate process-local caches. Concurrent requests +for the same ID share one promise. A successful name is reused on later +envelopes; a failed attempt is retained until the daemon restarts. Lookup HTTP, +API, parsing, and timeout failures produce no log output. + +When either lookup succeeds, Feishu writes a second observation through the +existing `ChannelBase` persistence method. The observation has the same +channel, user, group, and topic IDs, so `ObservedChannelContactStore` replaces +the ID labels without requiring a storage-format change. An unsuccessful +lookup leaves the first ID-based observation intact. + +## Ordering and access control + +Enrichment starts only after the existing inbound preflight and initial +observation succeed. Duplicate events, empty messages rejected by the adapter, +and messages rejected by sender or group policy do not trigger lookups. The +background lookup is not awaited by `handleInbound` or the agent prompt path. + +## Permissions + +Sender-name enrichment uses the least-privilege +`contact:user.basic_profile:readonly` scope. Group-name enrichment uses +`im:chat:readonly`. IDs remain application-scoped, so cross-application IDs and +external users may remain unresolved. + +## Testing + +Channel-base tests verify that the post-observation hook runs after preflight +and is not awaited. Feishu adapter tests verify successful enrichment, +process-local de-duplication, cached labels on later envelopes, and silent +failure while the original ID-based observation and inbound processing remain +available. From a9dd0039340189ad82c2596859f106a6d55862e5 Mon Sep 17 00:00:00 2001 From: hit_aran Date: Wed, 5 Aug 2026 14:56:28 +0800 Subject: [PATCH 2/7] docs: add Chinese Feishu enrichment design --- ...eishu-observed-contact-label-enrichment.md | 58 +++++++++++++++++-- 1 file changed, 53 insertions(+), 5 deletions(-) diff --git a/docs/design/feishu-observed-contact-label-enrichment.md b/docs/design/feishu-observed-contact-label-enrichment.md index 60b226c757a..65ce5ff7658 100644 --- a/docs/design/feishu-observed-contact-label-enrichment.md +++ b/docs/design/feishu-observed-contact-label-enrichment.md @@ -1,12 +1,60 @@ # Feishu observed-contact label enrichment -## Goal +## 中文说明 + +### 目标 + +在不延迟飞书入站消息处理的前提下，用真实用户名和群名补全观测联系人 +的 label。名称查询不可用时继续保留现有 ID label。 + +### 设计 + +`ChannelBase` 在入站消息通过 preflight 后，仍先立即落盘基于 ID 的观测 +记录。随后调用一个同步的 protected 后置观测钩子。默认实现不执行任何 +操作，返回值也不会被等待，因此不影响其他渠道。 + +`FeishuChannel` 覆写该钩子，对当前 channel 实例生命周期内尚未尝试过的 +ID 发起后台查询： + +- `POST /open-apis/contact/v3/users/basic_batch` 查询发送者姓名。 +- `GET /open-apis/im/v1/chats/:chat_id` 查询群名。 + +用户和群分别使用进程内缓存。同一个 ID 的并发请求复用同一个 Promise。 +查询成功的名称供后续消息直接复用；查询失败则记录到 daemon 重启为止。 +查询请求的 HTTP 错误、API 错误、解析错误和超时均不输出日志。 + +任一名称查询成功后，飞书通过 `ChannelBase` 现有持久化方法再次写入观测 +记录。由于 channel、用户、群和话题 ID 均未改变， +`ObservedChannelContactStore` 会直接用名称 label 替换 ID label，无需修改 +存储格式。查询均失败时，首次写入的 ID 观测记录保持不变。 + +### 顺序与访问控制 + +只有现有 preflight 和首次观测落盘成功后，才开始名称补全。重复事件、被 +适配器丢弃的空消息，以及未通过发送者或群策略的消息都不会触发查询。 +后台查询不会被 `handleInbound` 或 Agent prompt 主链路等待。 + +### 权限 + +发送者姓名查询使用最小权限 `contact:user.basic_profile:readonly`，群名查询 +使用 `im:chat:readonly`。ID 仍具有应用隔离性，因此跨应用 ID 和外部用户 +可能无法补全。 + +### 测试 + +Channel Base 测试验证后置观测钩子只在 preflight 后触发且不会被等待。 +飞书适配器测试验证成功补全、进程内去重、后续消息复用名称，以及失败静默 +时原始 ID 观测记录和入站消息处理仍然可用。 + +## English + +### Goal Populate Feishu observed-contact labels with the sender name and group name without delaying inbound message processing. Keep the current ID labels when lookup is unavailable. -## Design +### Design `ChannelBase` continues to persist the ID-based observation immediately after inbound preflight succeeds. It then invokes a synchronous, protected @@ -30,21 +78,21 @@ channel, user, group, and topic IDs, so `ObservedChannelContactStore` replaces the ID labels without requiring a storage-format change. An unsuccessful lookup leaves the first ID-based observation intact. -## Ordering and access control +### Ordering and access control Enrichment starts only after the existing inbound preflight and initial observation succeed. Duplicate events, empty messages rejected by the adapter, and messages rejected by sender or group policy do not trigger lookups. The background lookup is not awaited by `handleInbound` or the agent prompt path. -## Permissions +### Permissions Sender-name enrichment uses the least-privilege `contact:user.basic_profile:readonly` scope. Group-name enrichment uses `im:chat:readonly`. IDs remain application-scoped, so cross-application IDs and external users may remain unresolved. -## Testing +### Testing Channel-base tests verify that the post-observation hook runs after preflight and is not awaited. Feishu adapter tests verify successful enrichment, From c82d3f131b783bcb7126a95f7c754435f262cb5a Mon Sep 17 00:00:00 2001 From: hit_aran Date: Wed, 5 Aug 2026 15:36:12 +0800 Subject: [PATCH 3/7] feat(feishu): enrich observed contact labels --- ...eishu-observed-contact-label-enrichment.md | 11 +- ...eishu-observed-contact-label-enrichment.md | 290 ++++++++++++++++++ .../channels/base/src/ChannelBase.test.ts | 26 ++ packages/channels/base/src/ChannelBase.ts | 5 +- packages/channels/feishu/src/FeishuAdapter.ts | 118 ++++++- packages/channels/feishu/src/adapter.test.ts | 179 +++++++++++ 6 files changed, 622 insertions(+), 7 deletions(-) create mode 100644 docs/plans/feishu-observed-contact-label-enrichment.md diff --git a/docs/design/feishu-observed-contact-label-enrichment.md b/docs/design/feishu-observed-contact-label-enrichment.md index 65ce5ff7658..49fd09fd47a 100644 --- a/docs/design/feishu-observed-contact-label-enrichment.md +++ b/docs/design/feishu-observed-contact-label-enrichment.md @@ -30,7 +30,7 @@ ID 发起后台查询： ### 顺序与访问控制 -只有现有 preflight 和首次观测落盘成功后，才开始名称补全。重复事件、被 +只有现有 preflight 通过且首次观测落盘尝试完成后，才开始名称补全。重复事件、被 适配器丢弃的空消息，以及未通过发送者或群策略的消息都不会触发查询。 后台查询不会被 `handleInbound` 或 Agent prompt 主链路等待。 @@ -80,10 +80,11 @@ lookup leaves the first ID-based observation intact. ### Ordering and access control -Enrichment starts only after the existing inbound preflight and initial -observation succeed. Duplicate events, empty messages rejected by the adapter, -and messages rejected by sender or group policy do not trigger lookups. The -background lookup is not awaited by `handleInbound` or the agent prompt path. +Enrichment starts only after the existing inbound preflight succeeds and the +initial observation attempt completes. Duplicate events, empty messages +rejected by the adapter, and messages rejected by sender or group policy do not +trigger lookups. The background lookup is not awaited by `handleInbound` or the +agent prompt path. ### Permissions diff --git a/docs/plans/feishu-observed-contact-label-enrichment.md b/docs/plans/feishu-observed-contact-label-enrichment.md new file mode 100644 index 00000000000..793bf14b91f --- /dev/null +++ b/docs/plans/feishu-observed-contact-label-enrichment.md @@ -0,0 +1,290 @@ +# Feishu Observed-Contact Label Enrichment Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Resolve Feishu sender and group IDs into recognizable observed-contact labels without delaying inbound message processing. + +**Architecture:** `ChannelBase` exposes a synchronous post-observation hook and its existing protected persistence path. `FeishuChannel` starts cached, fire-and-forget OpenAPI lookups from that hook, then writes one enriched observation after successful resolution. The existing store schema and non-Feishu channels remain unchanged. + +**Tech Stack:** TypeScript, Node.js 22+, Vitest, Feishu tenant-access-token OpenAPI, existing channel-base observed-contact sink. + +## Global Constraints + +- Preserve the existing preflight boundary: rejected messages must not trigger lookups. +- Persist the ID observation before starting enrichment. +- Never await Feishu label lookup from `handleInbound` or the Agent prompt path. +- Query each user and group ID at most once per `FeishuChannel` instance; concurrent observations share the same promise. +- Cache failed attempts until daemon restart and emit no lookup-specific logs. +- Use `contact:user.basic_profile:readonly` and `im:chat:readonly`; do not use full-contact APIs. +- Keep ESM, strict TypeScript, no `any`, and existing file naming. + +--- + +### Task 1: Add the post-observation extension point + +**Files:** + +- Modify: `packages/channels/base/src/ChannelBase.ts` +- Test: `packages/channels/base/src/ChannelBase.test.ts` + +**Interfaces:** + +- Produces: `protected onObservedContact(envelope: Envelope): void` +- Produces: `protected recordObservedContact(envelope: Envelope): Promise` +- Guarantees: the hook runs once, after the initial persistence attempt, only for preflight-approved envelopes. + +- [ ] **Step 1: Write the failing hook-order test** + +Add a local test subclass and assert that persistence happens before the hook: + +```ts +class ObservedHookChannel extends TestChannel { + readonly observedEnvelopes: Envelope[] = []; + + protected override onObservedContact(envelope: Envelope): void { + this.observedEnvelopes.push(envelope); + } +} + +it('notifies the adapter after an approved contact is persisted', async () => { + const order: string[] = []; + const observe = vi.fn(() => { + order.push('persisted'); + }); + const ch = new ObservedHookChannel('test-chan', defaultConfig(), bridge, { + observedContacts: { observe }, + }); + const message = envelope(); + + await ch.handleInbound(message); + + expect(order).toEqual(['persisted']); + expect(ch.observedEnvelopes).toEqual([message]); + expect(bridge.prompt).toHaveBeenCalled(); +}); +``` + +- [ ] **Step 2: Run the base test and verify RED** + +Run: + +```bash +cd packages/channels/base +npx vitest run src/ChannelBase.test.ts -t "notifies the adapter after an approved contact is persisted" +``` + +Expected: compilation fails because `onObservedContact` does not exist. + +- [ ] **Step 3: Implement the minimal base hook** + +Change only the persistence method visibility from `private` to `protected`; keep its body unchanged. Add the default no-op hook and invoke it synchronously after the initial observation: + +```ts +protected async recordObservedContact(envelope: Envelope): Promise { +``` + +```ts +protected onObservedContact(_envelope: Envelope): void {} + +// Inside processInbound, after await this.recordObservedContact(envelope): +this.onObservedContact(envelope); +``` + +- [ ] **Step 4: Run the base test and verify GREEN** + +Run: + +```bash +cd packages/channels/base +npx vitest run src/ChannelBase.test.ts +``` + +Expected: all ChannelBase tests pass. + +### Task 2: Enrich Feishu observations in the background + +**Files:** + +- Modify: `packages/channels/feishu/src/FeishuAdapter.ts` +- Test: `packages/channels/feishu/src/adapter.test.ts` + +**Interfaces:** + +- Consumes: `onObservedContact(envelope)` and `recordObservedContact(envelope)` from Task 1. +- Produces: process-local user/group name and lookup-promise caches. +- Produces: `basic_batch` requests with the callback ID type and `GET /im/v1/chats/:chat_id` requests. + +- [ ] **Step 1: Write failing success and cache tests** + +Allow the test factory to pass `ChannelBaseOptions`, preload `tokenCache`, and send two complete Feishu group-event fixtures with different message IDs. Mock these complete API responses: + +```ts +new Response( + JSON.stringify({ + code: 0, + msg: 'success', + data: { users: [{ user_id: 'ou_user', name: 'Alice' }] }, + }), + { status: 200 }, +); + +new Response( + JSON.stringify({ + code: 0, + msg: 'success', + data: { name: 'Project Group' }, + }), + { status: 200 }, +); +``` + +Assert the observable store calls, not private maps: + +```ts +expect(observe).toHaveBeenNthCalledWith(1, 'test', { + user: { id: 'ou_user', label: 'ou_user' }, + group: { id: 'oc_group', label: 'oc_group' }, +}); +expect(observe).toHaveBeenNthCalledWith(2, 'test', { + user: { id: 'ou_user', label: 'Alice' }, + group: { id: 'oc_group', label: 'Project Group' }, +}); +``` + +After the first lookup settles, deliver the second fixture and assert its first observation already contains the cached names and that each OpenAPI endpoint was requested exactly once. + +- [ ] **Step 2: Write the failing non-blocking and silent-failure test** + +Use an externally controlled rejected lookup promise, a real mock bridge, and a no-op `onPromptStart` test subclass. Assert `bridge.prompt` runs while lookup work is unresolved. After rejecting the lookups, assert the ID observation remains and `process.stderr.write` received no lookup error. + +- [ ] **Step 3: Run Feishu tests and verify RED** + +Run: + +```bash +cd packages/channels/feishu +npx vitest run src/adapter.test.ts -t "observed contact" +``` + +Expected: the enriched observation and background request assertions fail because Feishu does not yet resolve labels. + +- [ ] **Step 4: Implement cached background lookups** + +Add successful-name maps and promise caches: + +```ts +private readonly observedUserNames = new Map(); +private readonly observedChatNames = new Map(); +private readonly observedUserLookups = new Map< + string, + Promise +>(); +private readonly observedChatLookups = new Map< + string, + Promise +>(); +``` + +Populate later envelopes synchronously from successful caches: + +```ts +const cachedSenderName = this.observedUserNames.get(senderId); +const cachedChatName = isGroup ? this.observedChatNames.get(chatId) : undefined; + +const envelope: Envelope = { + channelName: this.name, + senderId, + senderName: cachedSenderName || senderId, + chatId, + text: cleanText, + messageId: msgId, + threadId: msg.root_id || undefined, + isGroup, + isMentioned, + isReplyToBot: false, + ...(cachedChatName ? { chatName: cachedChatName } : {}), +}; +``` + +Override the base hook without returning or awaiting the lookup promise: + +```ts +protected override onObservedContact(envelope: Envelope): void { + void this.enrichObservedContact(envelope); +} +``` + +Implement user and chat request helpers with `AbortSignal.timeout(15_000)`. Convert `ou_` to `open_id`, `on_` to `union_id`, and other IDs to `user_id`. Return `undefined` for missing tokens, non-2xx responses, non-zero API codes, malformed JSON, empty names, and thrown errors without writing logs. Store the promise before awaiting it so concurrent messages share the same request; retain resolved `undefined` promises to suppress retries until restart. + +After `Promise.all`, call: + +```ts +await this.recordObservedContact({ + ...envelope, + ...(senderName ? { senderName } : {}), + ...(chatName ? { chatName } : {}), +}); +``` + +only when at least one lookup returned a name. + +- [ ] **Step 5: Run Feishu tests and verify GREEN** + +Run: + +```bash +cd packages/channels/feishu +npx vitest run src/adapter.test.ts +``` + +Expected: all Feishu adapter tests pass with no unexpected output from the new failure cases. + +### Task 3: Verify, review, and publish + +**Files:** + +- Verify: `packages/channels/base/src/ChannelBase.ts` +- Verify: `packages/channels/base/src/ChannelBase.test.ts` +- Verify: `packages/channels/feishu/src/FeishuAdapter.ts` +- Verify: `packages/channels/feishu/src/adapter.test.ts` +- Include: `docs/design/feishu-observed-contact-label-enrichment.md` +- Include: `docs/plans/feishu-observed-contact-label-enrichment.md` + +**Interfaces:** + +- Links: GitHub issue `QwenLM/qwen-code#8566`. +- Produces: one draft pull request from `BenGuanRan:feat/feishu-observed-contact-labels` to `QwenLM/qwen-code:main`. + +- [ ] **Step 1: Run focused verification** + +```bash +cd packages/channels/base && npx vitest run src/ChannelBase.test.ts +cd ../feishu && npx vitest run src/adapter.test.ts +cd ../../.. && npm run build && npm run typecheck +git diff --check origin/main...HEAD +``` + +- [ ] **Step 2: Self-audit the complete diff** + +Read `git diff origin/main...HEAD` in open-ended passes. Check every changed behavior against the design, verify the tests would fail if the hook or enrichment write were removed, and stop after two consecutive clean passes. + +- [ ] **Step 3: Run the Codex review workflow** + +Review the exact branch diff against `origin/main`. Triage every finding as valid, false positive, or overthinking; accepted fixes return to focused tests and self-audit. + +- [ ] **Step 4: Commit after explicit staging and commit authorization** + +```bash +git add -- packages/channels/base/src/ChannelBase.ts packages/channels/base/src/ChannelBase.test.ts packages/channels/feishu/src/FeishuAdapter.ts packages/channels/feishu/src/adapter.test.ts docs/plans/feishu-observed-contact-label-enrichment.md +git commit -m "feat(feishu): enrich observed contact labels" +``` + +- [ ] **Step 5: Push after explicit push authorization** + +```bash +git push -u fork feat/feishu-observed-contact-labels +``` + +- [ ] **Step 6: Create the authorized draft PR with the repository template** + +Create exactly one draft PR with title `feat(feishu): enrich observed contact labels`, base `QwenLM/qwen-code:main`, head `BenGuanRan:feat/feishu-observed-contact-labels`, and `Closes #8566`. Fill every English template section and provide a complete paragraph-for-paragraph Chinese translation in the `
` block. diff --git a/packages/channels/base/src/ChannelBase.test.ts b/packages/channels/base/src/ChannelBase.test.ts index 1799d109968..cea902c5a57 100644 --- a/packages/channels/base/src/ChannelBase.test.ts +++ b/packages/channels/base/src/ChannelBase.test.ts @@ -907,6 +907,32 @@ describe('ChannelBase', () => { expect(bridge.prompt).toHaveBeenCalled(); }); + it('notifies the adapter after an approved contact is persisted', async () => { + const order: string[] = []; + class ObservedHookChannel extends TestChannel { + readonly observedEnvelopes: Envelope[] = []; + + protected override onObservedContact(envelope: Envelope): void { + order.push('hook'); + this.observedEnvelopes.push(envelope); + } + } + + const observe = vi.fn(() => { + order.push('persisted'); + }); + const ch = new ObservedHookChannel('test-chan', defaultConfig(), bridge, { + observedContacts: { observe }, + }); + const message = envelope(); + + await ch.handleInbound(message); + + expect(order).toEqual(['persisted', 'hook']); + expect(ch.observedEnvelopes).toEqual([message]); + expect(bridge.prompt).toHaveBeenCalled(); + }); + it('falls back to the complete sender ID for an unusable label', async () => { const observe = vi.fn(); const ch = createChannel({}, { observedContacts: { observe } }); diff --git a/packages/channels/base/src/ChannelBase.ts b/packages/channels/base/src/ChannelBase.ts index 10450bcf359..38b39876227 100644 --- a/packages/channels/base/src/ChannelBase.ts +++ b/packages/channels/base/src/ChannelBase.ts @@ -4876,7 +4876,7 @@ export abstract class ChannelBase { await this.processInbound(envelope); } - private async recordObservedContact(envelope: Envelope): Promise { + protected async recordObservedContact(envelope: Envelope): Promise { if (!this.observedContacts) return; const sanitizedSenderName = envelope.senderName ? sanitizeSenderName(envelope.senderName) @@ -4917,6 +4917,8 @@ export abstract class ChannelBase { } } + protected onObservedContact(_envelope: Envelope): void {} + protected markPreflighted(envelope: Envelope): void { this.preflightedEnvelopes.add(envelope); } @@ -4949,6 +4951,7 @@ export abstract class ChannelBase { if (this.observedContacts && !this.observedContactEnvelopes.has(envelope)) { this.observedContactEnvelopes.add(envelope); await this.recordObservedContact(envelope); + this.onObservedContact(envelope); } let memoryIntent: ResolvedChannelMemoryIntent | null = diff --git a/packages/channels/feishu/src/FeishuAdapter.ts b/packages/channels/feishu/src/FeishuAdapter.ts index 92b7aaf4ca6..68c45395df5 100644 --- a/packages/channels/feishu/src/FeishuAdapter.ts +++ b/packages/channels/feishu/src/FeishuAdapter.ts @@ -142,6 +142,16 @@ export class FeishuChannel extends ChannelBase { private botOpenId?: string; private tokenCache?: { token: string; expiresAt: number }; private tokenRefreshPromise?: Promise; + private readonly observedUserNames = new Map(); + private readonly observedChatNames = new Map(); + private readonly observedUserLookups = new Map< + string, + Promise + >(); + private readonly observedChatLookups = new Map< + string, + Promise + >(); private collapsible: boolean; private collapsibleThreshold: number; @@ -676,6 +686,109 @@ export class FeishuChannel extends ChannelBase { } } + private observedUserName(userId: string): Promise { + const existing = this.observedUserLookups.get(userId); + if (existing) return existing; + + const lookup = (async () => { + try { + const token = await this.getTenantAccessToken(); + if (!token) return undefined; + + const userIdType = userId.startsWith('ou_') + ? 'open_id' + : userId.startsWith('on_') + ? 'union_id' + : 'user_id'; + const response = await fetch( + `${BASE_URL}/contact/v3/users/basic_batch?user_id_type=${userIdType}`, + { + method: 'POST', + headers: { + Authorization: `Bearer ${token}`, + 'Content-Type': 'application/json', + }, + body: JSON.stringify({ user_ids: [userId] }), + signal: AbortSignal.timeout(15_000), + }, + ); + if (!response.ok) return undefined; + + const body = (await response.json()) as { + code?: number; + data?: { users?: Array<{ name?: string }> }; + }; + const name = body.data?.users?.[0]?.name?.trim(); + if (body.code !== 0 || !name) return undefined; + this.observedUserNames.set(userId, name); + return name; + } catch { + return undefined; + } + })(); + this.observedUserLookups.set(userId, lookup); + return lookup; + } + + private observedChatName(chatId: string): Promise { + const existing = this.observedChatLookups.get(chatId); + if (existing) return existing; + + const lookup = (async () => { + try { + const token = await this.getTenantAccessToken(); + if (!token) return undefined; + + const response = await fetch( + `${BASE_URL}/im/v1/chats/${encodeURIComponent(chatId)}`, + { + headers: { Authorization: `Bearer ${token}` }, + signal: AbortSignal.timeout(15_000), + }, + ); + if (!response.ok) return undefined; + + const body = (await response.json()) as { + code?: number; + data?: { name?: string }; + }; + const name = body.data?.name?.trim(); + if (body.code !== 0 || !name) return undefined; + this.observedChatNames.set(chatId, name); + return name; + } catch { + return undefined; + } + })(); + this.observedChatLookups.set(chatId, lookup); + return lookup; + } + + protected override onObservedContact(envelope: Envelope): void { + void this.enrichObservedContact(envelope).catch(() => {}); + } + + private async enrichObservedContact(envelope: Envelope): Promise { + const userAttempted = this.observedUserLookups.has(envelope.senderId); + const chatAttempted = + !envelope.isGroup || this.observedChatLookups.has(envelope.chatId); + if (userAttempted && chatAttempted) return; + + const [senderName, chatName] = await Promise.all([ + this.observedUserName(envelope.senderId), + envelope.isGroup + ? this.observedChatName(envelope.chatId) + : Promise.resolve(undefined), + ]); + if (!senderName && !chatName) return; + + await this.recordObservedContact({ + ...envelope, + ...(senderName ? { senderName } : {}), + ...(chatName ? { chatName } : {}), + }); + } + async sendMessage(chatId: string, text: string): Promise { await this.sendMessageInternal(chatId, text, false); } @@ -1951,6 +2064,8 @@ export class FeishuChannel extends ChannelBase { sender.sender_id?.user_id || sender.sender_id?.union_id || ''; + const senderName = this.observedUserNames.get(senderId) || senderId; + const chatName = isGroup ? this.observedChatNames.get(chatId) : undefined; // Parse message content const content = this.extractContent(msg.message_type, msg.content); @@ -1995,8 +2110,9 @@ export class FeishuChannel extends ChannelBase { const envelope: Envelope = { channelName: this.name, senderId, - senderName: senderId, + senderName, chatId, + ...(chatName ? { chatName } : {}), text: cleanText, messageId: msgId, threadId: msg.root_id || undefined, diff --git a/packages/channels/feishu/src/adapter.test.ts b/packages/channels/feishu/src/adapter.test.ts index 97de4df1f08..6510fbb0173 100644 --- a/packages/channels/feishu/src/adapter.test.ts +++ b/packages/channels/feishu/src/adapter.test.ts @@ -20,6 +20,7 @@ vi.mock('@larksuiteoapi/node-sdk', async (importOriginal) => { import { FeishuChannel } from './FeishuAdapter.js'; import type { ChannelAgentBridge, + ChannelBaseOptions, ChannelConfig, ChannelProactiveDeliveryError, ChannelTaskLifecycleEvent, @@ -77,6 +78,61 @@ function createTestableChannel( return new TestableFeishuChannel('test', config, bridge); } +class ObservedContactFeishuChannel extends FeishuChannel { + protected override onPromptStart(): void {} + + protected override onPromptEnd(): Promise { + return Promise.resolve(); + } +} + +function createObservedContactChannel( + observe: NonNullable['observe'], +): { + channel: ObservedContactFeishuChannel; + bridge: ChannelAgentBridge; +} { + const bridge = createMockBridge(); + const channel = new ObservedContactFeishuChannel( + 'test', + createConfig({ + blockStreaming: 'on', + groupPolicy: 'open', + groups: { '*': { requireMention: false } }, + }), + bridge, + { observedContacts: { observe } }, + ); + Object.assign(channel as unknown as Record, { + tokenCache: { + token: 'test_token', + expiresAt: Date.now() + 3_600_000, + }, + }); + return { channel, bridge }; +} + +function feishuGroupMessage(messageId: string): Record { + return { + message: { + message_id: messageId, + chat_id: 'oc_group', + chat_type: 'group', + message_type: 'text', + content: JSON.stringify({ text: 'hello' }), + }, + sender: { + sender_id: { + union_id: 'on_user', + user_id: 'user_1', + open_id: 'ou_user', + }, + sender_type: 'user', + tenant_key: 'tenant_1', + }, + }; +} + // Access private methods for unit testing function getPrivateMethod(instance: unknown, method: string): T { return (instance as Record)[method] as T; @@ -259,6 +315,129 @@ describe('FeishuChannel', () => { }); }); + describe('observed contact enrichment', () => { + it('resolves labels once and reuses them on later observations', async () => { + const observe = vi.fn(); + const { channel } = createObservedContactChannel(observe); + const fetchSpy = vi + .spyOn(global, 'fetch') + .mockImplementation(async (input) => { + const url = String(input); + if (url.includes('/contact/v3/users/basic_batch')) { + return new Response( + JSON.stringify({ + code: 0, + msg: 'success', + data: { + users: [{ user_id: 'ou_user', name: 'Alice' }], + }, + }), + { status: 200 }, + ); + } + if (url.includes('/im/v1/chats/oc_group')) { + return new Response( + JSON.stringify({ + code: 0, + msg: 'success', + data: { name: 'Project Group' }, + }), + { status: 200 }, + ); + } + throw new Error(`Unexpected request: ${url}`); + }); + const onMessage = getPrivateMethod<(data: unknown) => void>( + channel, + 'onMessage', + ).bind(channel); + + try { + onMessage(feishuGroupMessage('message_1')); + + await vi.waitFor(() => expect(observe).toHaveBeenCalledTimes(2)); + expect(observe).toHaveBeenNthCalledWith(1, 'test', { + user: { id: 'ou_user', label: 'ou_user' }, + group: { id: 'oc_group', label: 'oc_group' }, + }); + expect(observe).toHaveBeenNthCalledWith(2, 'test', { + user: { id: 'ou_user', label: 'Alice' }, + group: { id: 'oc_group', label: 'Project Group' }, + }); + expect(fetchSpy).toHaveBeenCalledWith( + expect.stringContaining('user_id_type=open_id'), + expect.objectContaining({ + method: 'POST', + body: JSON.stringify({ user_ids: ['ou_user'] }), + }), + ); + expect(fetchSpy).toHaveBeenCalledWith( + expect.stringContaining('/im/v1/chats/oc_group'), + expect.objectContaining({ + headers: expect.objectContaining({ + Authorization: 'Bearer test_token', + }), + }), + ); + + onMessage(feishuGroupMessage('message_2')); + + await vi.waitFor(() => expect(observe).toHaveBeenCalledTimes(3)); + expect(observe).toHaveBeenNthCalledWith(3, 'test', { + user: { id: 'ou_user', label: 'Alice' }, + group: { id: 'oc_group', label: 'Project Group' }, + }); + expect(fetchSpy).toHaveBeenCalledTimes(2); + } finally { + fetchSpy.mockRestore(); + } + }); + + it('does not block inbound processing and silently caches lookup failures', async () => { + const observe = vi.fn(); + const { channel, bridge } = createObservedContactChannel(observe); + let rejectLookup: (reason: Error) => void = () => {}; + const lookup = new Promise((_resolve, reject) => { + rejectLookup = reject; + }); + const fetchSpy = vi + .spyOn(global, 'fetch') + .mockImplementation(() => lookup); + const stderrSpy = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + const onMessage = getPrivateMethod<(data: unknown) => void>( + channel, + 'onMessage', + ).bind(channel); + + try { + onMessage(feishuGroupMessage('message_1')); + + await vi.waitFor(() => { + expect(observe).toHaveBeenCalledTimes(1); + expect(bridge.prompt).toHaveBeenCalledTimes(1); + }); + expect(fetchSpy).toHaveBeenCalledTimes(2); + + rejectLookup(new Error('private lookup failure')); + await new Promise((resolve) => setImmediate(resolve)); + + onMessage(feishuGroupMessage('message_2')); + await vi.waitFor(() => { + expect(observe).toHaveBeenCalledTimes(2); + expect(bridge.prompt).toHaveBeenCalledTimes(2); + }); + + expect(fetchSpy).toHaveBeenCalledTimes(2); + expect(stderrSpy).not.toHaveBeenCalled(); + } finally { + fetchSpy.mockRestore(); + stderrSpy.mockRestore(); + } + }); + }); + describe('extractCardText', () => { let channel: FeishuChannel; let extractCardText: (card: Record) => string | undefined; From 8ef1b1e35cf03082ed2ddbc92fb381c3f95cf189 Mon Sep 17 00:00:00 2001 From: hit_aran Date: Thu, 6 Aug 2026 15:04:49 +0800 Subject: [PATCH 4/7] fix(feishu): preserve enriched contact labels --- docs/users/features/channels/feishu.md | 9 + packages/channels/feishu/src/FeishuAdapter.ts | 31 ++- packages/channels/feishu/src/adapter.test.ts | 178 ++++++++++++++++++ 3 files changed, 209 insertions(+), 9 deletions(-) diff --git a/docs/users/features/channels/feishu.md b/docs/users/features/channels/feishu.md index 660d3b0e312..0ac3a974958 100644 --- a/docs/users/features/channels/feishu.md +++ b/docs/users/features/channels/feishu.md @@ -38,6 +38,15 @@ Enable the following permissions under **Permissions & Scopes** (权限管理): - `im:message:send_as_bot` — Send messages as bot - `im:resource` — Access message resources (images, files) +To show user and group names instead of IDs in daemon-discovered contacts, +optionally enable: + +- `contact:user.basic_profile:readonly` — Read user display names +- `im:chat:readonly` — Read group names + +Without these optional permissions, messages still work and discovered contacts +keep their Feishu user and chat IDs as labels. + ### Publish the Application After configuring permissions and events, create a version and publish it. The bot won't work until the application is published and approved. diff --git a/packages/channels/feishu/src/FeishuAdapter.ts b/packages/channels/feishu/src/FeishuAdapter.ts index 68c45395df5..6d622c625ef 100644 --- a/packages/channels/feishu/src/FeishuAdapter.ts +++ b/packages/channels/feishu/src/FeishuAdapter.ts @@ -693,7 +693,10 @@ export class FeishuChannel extends ChannelBase { const lookup = (async () => { try { const token = await this.getTenantAccessToken(); - if (!token) return undefined; + if (!token) { + this.observedUserLookups.delete(userId); + return undefined; + } const userIdType = userId.startsWith('ou_') ? 'open_id' @@ -712,7 +715,10 @@ export class FeishuChannel extends ChannelBase { signal: AbortSignal.timeout(15_000), }, ); - if (!response.ok) return undefined; + if (!response.ok) { + if (response.status === 401) this.tokenCache = undefined; + return undefined; + } const body = (await response.json()) as { code?: number; @@ -737,7 +743,10 @@ export class FeishuChannel extends ChannelBase { const lookup = (async () => { try { const token = await this.getTenantAccessToken(); - if (!token) return undefined; + if (!token) { + this.observedChatLookups.delete(chatId); + return undefined; + } const response = await fetch( `${BASE_URL}/im/v1/chats/${encodeURIComponent(chatId)}`, @@ -746,7 +755,10 @@ export class FeishuChannel extends ChannelBase { signal: AbortSignal.timeout(15_000), }, ); - if (!response.ok) return undefined; + if (!response.ok) { + if (response.status === 401) this.tokenCache = undefined; + return undefined; + } const body = (await response.json()) as { code?: number; @@ -769,11 +781,6 @@ export class FeishuChannel extends ChannelBase { } private async enrichObservedContact(envelope: Envelope): Promise { - const userAttempted = this.observedUserLookups.has(envelope.senderId); - const chatAttempted = - !envelope.isGroup || this.observedChatLookups.has(envelope.chatId); - if (userAttempted && chatAttempted) return; - const [senderName, chatName] = await Promise.all([ this.observedUserName(envelope.senderId), envelope.isGroup @@ -781,6 +788,12 @@ export class FeishuChannel extends ChannelBase { : Promise.resolve(undefined), ]); if (!senderName && !chatName) return; + if ( + (!senderName || envelope.senderName === senderName) && + (!chatName || envelope.chatName === chatName) + ) { + return; + } await this.recordObservedContact({ ...envelope, diff --git a/packages/channels/feishu/src/adapter.test.ts b/packages/channels/feishu/src/adapter.test.ts index 6510fbb0173..aa9d58dd599 100644 --- a/packages/channels/feishu/src/adapter.test.ts +++ b/packages/channels/feishu/src/adapter.test.ts @@ -138,6 +138,23 @@ function getPrivateMethod(instance: unknown, method: string): T { return (instance as Record)[method] as T; } +function deferred(): { + promise: Promise; + resolve: (value: T) => void; +} { + let resolve: (value: T) => void = () => {}; + const promise = new Promise((resolvePromise) => { + resolve = resolvePromise; + }); + return { promise, resolve }; +} + +function jsonResponse(body: unknown, status = 200): Response { + return new Response(typeof body === 'string' ? body : JSON.stringify(body), { + status, + }); +} + describe('FeishuChannel', () => { describe('constructor', () => { it('throws if clientId is missing', () => { @@ -436,6 +453,167 @@ describe('FeishuChannel', () => { stderrSpy.mockRestore(); } }); + + it('repairs labels from a delayed message with a stale snapshot', async () => { + const observe = vi.fn(); + const { channel } = createObservedContactChannel(observe); + const userLookup = deferred(); + const chatLookup = deferred(); + const quotedMessage = deferred(); + const fetchSpy = vi.spyOn(global, 'fetch').mockImplementation((input) => { + const url = String(input); + if (url.includes('/contact/v3/users/basic_batch')) { + return userLookup.promise; + } + if (url.includes('/im/v1/chats/oc_group')) { + return chatLookup.promise; + } + if (url.includes('/im/v1/messages/om_parent')) { + return quotedMessage.promise; + } + throw new Error(`Unexpected request: ${url}`); + }); + const onMessage = getPrivateMethod<(data: unknown) => void>( + channel, + 'onMessage', + ).bind(channel); + + try { + onMessage(feishuGroupMessage('message_1')); + await vi.waitFor(() => expect(fetchSpy).toHaveBeenCalledTimes(2)); + + const reply = feishuGroupMessage('message_2'); + (reply['message'] as Record)['parent_id'] = + 'om_parent'; + onMessage(reply); + await vi.waitFor(() => expect(fetchSpy).toHaveBeenCalledTimes(3)); + + userLookup.resolve( + jsonResponse({ + code: 0, + data: { users: [{ name: 'Alice' }] }, + }), + ); + chatLookup.resolve( + jsonResponse({ code: 0, data: { name: 'Project Group' } }), + ); + await vi.waitFor(() => expect(observe).toHaveBeenCalledTimes(2)); + + quotedMessage.resolve( + jsonResponse({ + data: { + items: [ + { + msg_type: 'text', + body: { content: JSON.stringify({ text: 'quoted' }) }, + sender: { sender_type: 'user' }, + }, + ], + }, + }), + ); + + await vi.waitFor(() => expect(observe).toHaveBeenCalledTimes(4)); + expect(observe).toHaveBeenNthCalledWith(3, 'test', { + user: { id: 'ou_user', label: 'ou_user' }, + group: { id: 'oc_group', label: 'oc_group' }, + }); + expect(observe).toHaveBeenLastCalledWith('test', { + user: { id: 'ou_user', label: 'Alice' }, + group: { id: 'oc_group', label: 'Project Group' }, + }); + expect( + fetchSpy.mock.calls.filter(([input]) => + String(input).includes('/contact/v3/users/basic_batch'), + ), + ).toHaveLength(1); + expect( + fetchSpy.mock.calls.filter(([input]) => + String(input).includes('/im/v1/chats/oc_group'), + ), + ).toHaveLength(1); + } finally { + fetchSpy.mockRestore(); + } + }); + + it('retries a label lookup when token acquisition fails before the request', async () => { + const observe = vi.fn(); + const { channel } = createObservedContactChannel(observe); + Object.assign(channel as unknown as Record, { + tokenCache: undefined, + }); + const fetchSpy = vi + .spyOn(global, 'fetch') + .mockResolvedValueOnce(jsonResponse('unavailable', 503)) + .mockResolvedValueOnce( + jsonResponse({ + tenant_access_token: 'fresh_token', + expire: 3600, + }), + ) + .mockResolvedValueOnce( + jsonResponse({ + code: 0, + data: { users: [{ name: 'Alice' }] }, + }), + ); + const stderrSpy = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + const observedUserName = getPrivateMethod< + (userId: string) => Promise + >(channel, 'observedUserName').bind(channel); + + try { + await expect(observedUserName('ou_user')).resolves.toBeUndefined(); + await expect(observedUserName('ou_user')).resolves.toBe('Alice'); + expect(fetchSpy).toHaveBeenCalledTimes(3); + } finally { + fetchSpy.mockRestore(); + stderrSpy.mockRestore(); + } + }); + + it('refreshes the token after a label lookup returns 401', async () => { + const observe = vi.fn(); + const { channel } = createObservedContactChannel(observe); + const fetchSpy = vi + .spyOn(global, 'fetch') + .mockImplementation(async (input, init) => { + const url = String(input); + if (url.includes('/tenant_access_token/internal')) { + return jsonResponse({ + tenant_access_token: 'fresh_token', + expire: 3600, + }); + } + if (url.includes('/contact/v3/users/basic_batch')) { + const authorization = new Headers(init?.headers).get( + 'Authorization', + ); + if (authorization === 'Bearer test_token') { + return jsonResponse('unauthorized', 401); + } + return jsonResponse({ + code: 0, + data: { users: [{ name: 'Bob' }] }, + }); + } + throw new Error(`Unexpected request: ${url}`); + }); + const observedUserName = getPrivateMethod< + (userId: string) => Promise + >(channel, 'observedUserName').bind(channel); + + try { + await expect(observedUserName('ou_first')).resolves.toBeUndefined(); + await expect(observedUserName('ou_second')).resolves.toBe('Bob'); + expect(fetchSpy).toHaveBeenCalledTimes(3); + } finally { + fetchSpy.mockRestore(); + } + }); }); describe('extractCardText', () => { From d2dec013a62c658425093098bfbdfce24423018d Mon Sep 17 00:00:00 2001 From: hit_aran Date: Thu, 6 Aug 2026 14:32:37 +0000 Subject: [PATCH 5/7] fix(feishu): harden observed-contact label enrichment lifecycle --- ...eishu-observed-contact-label-enrichment.md | 36 +- packages/channels/base/src/ChannelBase.ts | 24 + packages/channels/feishu/src/FeishuAdapter.ts | 162 +++--- packages/channels/feishu/src/adapter.test.ts | 474 +++++++++++++++++- .../commands/channel/daemon-worker.test.ts | 9 + .../cli/src/commands/channel/daemon-worker.ts | 9 +- 6 files changed, 644 insertions(+), 70 deletions(-) diff --git a/docs/design/feishu-observed-contact-label-enrichment.md b/docs/design/feishu-observed-contact-label-enrichment.md index 49fd09fd47a..fd37444f76b 100644 --- a/docs/design/feishu-observed-contact-label-enrichment.md +++ b/docs/design/feishu-observed-contact-label-enrichment.md @@ -19,9 +19,14 @@ ID 发起后台查询： - `POST /open-apis/contact/v3/users/basic_batch` 查询发送者姓名。 - `GET /open-apis/im/v1/chats/:chat_id` 查询群名。 -用户和群分别使用进程内缓存。同一个 ID 的并发请求复用同一个 Promise。 -查询成功的名称供后续消息直接复用；查询失败则记录到 daemon 重启为止。 -查询请求的 HTTP 错误、API 错误、解析错误和超时均不输出日志。 +用户和群分别使用进程内缓存，但共用同一套查询生命周期：同一个 ID +的并发请求复用同一个 Promise。查询成功的名称先经过共享的发送者名称净 +化，再供后续消息直接复用；请求已送达飞书的失败查询记录到 daemon 重 +启为止，因 tenant token 获取失败而未能发出的请求以及收到 401 的请求 +保持可重试，401 同时使缓存的 tenant token 失效。daemon 重启后第一条 +入站消息会从已持久化的观测联系人记录回填此前解析出的名称，避免已知 +名称在重新查询期间被回退为原始 ID。查询请求的 HTTP 错误、API 错误、 +解析错误和超时均不输出日志。 任一名称查询成功后，飞书通过 `ChannelBase` 现有持久化方法再次写入观测 记录。由于 channel、用户、群和话题 ID 均未改变， @@ -43,8 +48,8 @@ ID 发起后台查询： ### 测试 Channel Base 测试验证后置观测钩子只在 preflight 后触发且不会被等待。 -飞书适配器测试验证成功补全、进程内去重、后续消息复用名称，以及失败静默 -时原始 ID 观测记录和入站消息处理仍然可用。 +飞书适配器测试验证成功补全、进程内去重、后续消息复用名称、daemon 重启 +后回填已解析名称，以及失败静默时原始 ID 观测记录和入站消息处理仍然可用。 ## English @@ -67,10 +72,17 @@ have not been attempted during the current channel instance lifetime: - `POST /open-apis/contact/v3/users/basic_batch` resolves the sender name. - `GET /open-apis/im/v1/chats/:chat_id` resolves the group name. -User and group lookups have separate process-local caches. Concurrent requests -for the same ID share one promise. A successful name is reused on later -envelopes; a failed attempt is retained until the daemon restarts. Lookup HTTP, -API, parsing, and timeout failures produce no log output. +User and group lookups share one lookup lifecycle: process-local caches dedupe +by ID, and concurrent requests for the same ID share one promise. A successful +name is sanitized with the shared sender-name sanitizer and reused on later +envelopes. A failed attempt whose request reached Feishu is retained until the +daemon restarts; attempts never issued because tenant-token acquisition failed, +and requests answered with 401, remain retryable, and a 401 also invalidates +the cached tenant token. The first inbound message after a daemon restart +hydrates labels resolved by earlier runs from the persisted observed-contact +registry, so known names are not reverted to raw IDs while the fresh lookup is +pending. Lookup HTTP, API, parsing, and timeout failures produce no log +output. When either lookup succeeds, Feishu writes a second observation through the existing `ChannelBase` persistence method. The observation has the same @@ -97,6 +109,6 @@ external users may remain unresolved. Channel-base tests verify that the post-observation hook runs after preflight and is not awaited. Feishu adapter tests verify successful enrichment, -process-local de-duplication, cached labels on later envelopes, and silent -failure while the original ID-based observation and inbound processing remain -available. +process-local de-duplication, cached labels on later envelopes, hydration of +previously resolved labels after a daemon restart, and silent failure while the +original ID-based observation and inbound processing remain available. diff --git a/packages/channels/base/src/ChannelBase.ts b/packages/channels/base/src/ChannelBase.ts index 38b39876227..44818b68e9a 100644 --- a/packages/channels/base/src/ChannelBase.ts +++ b/packages/channels/base/src/ChannelBase.ts @@ -20,6 +20,7 @@ import type { ChannelUserQuestion, DispatchMode, Envelope, + ObservedChannelContactGraph, ObservedChannelContactObservation, SanitizedToolCallEvent, SessionTarget, @@ -229,6 +230,8 @@ export interface ChannelBaseOptions { channelName: string, observation: ObservedChannelContactObservation, ): void | Promise; + /** Read persisted observations so adapters can hydrate label caches. */ + list?(): ObservedChannelContactGraph; }; } @@ -4919,6 +4922,27 @@ export abstract class ChannelBase { protected onObservedContact(_envelope: Envelope): void {} + /** + * Observations persisted for this channel, when a read path is configured. + * Adapters hydrate label caches from it after a restart so known labels are + * not reverted to raw IDs by the next initial write. + */ + protected persistedObservedContacts(): + | ObservedChannelContactGraph + | undefined { + const list = this.observedContacts?.list; + if (!list) return undefined; + try { + const graph = list(); + return { + users: graph.users.filter((user) => user.channelName === this.name), + groups: graph.groups.filter((group) => group.channelName === this.name), + }; + } catch { + return undefined; + } + } + protected markPreflighted(envelope: Envelope): void { this.preflightedEnvelopes.add(envelope); } diff --git a/packages/channels/feishu/src/FeishuAdapter.ts b/packages/channels/feishu/src/FeishuAdapter.ts index 6d622c625ef..0ae90855e32 100644 --- a/packages/channels/feishu/src/FeishuAdapter.ts +++ b/packages/channels/feishu/src/FeishuAdapter.ts @@ -10,6 +10,7 @@ import { ChannelProactiveDeliveryError, isChannelProactiveDeliveryError, isTerminalTaskLifecycleType, + sanitizeSenderName, } from '@qwen-code/channel-base'; import { buildCardContent, extractTitle, splitChunks } from './markdown.js'; import { downloadMedia } from './media.js'; @@ -152,6 +153,11 @@ export class FeishuChannel extends ChannelBase { string, Promise >(); + private readonly observedContactWrites = new Map< + string, + { senderName: string; chatName: string | undefined } + >(); + private hydratedObservedNames = false; private collapsible: boolean; private collapsibleThreshold: number; @@ -686,24 +692,41 @@ export class FeishuChannel extends ChannelBase { } } - private observedUserName(userId: string): Promise { - const existing = this.observedUserLookups.get(userId); - if (existing) return existing; - - const lookup = (async () => { - try { - const token = await this.getTenantAccessToken(); - if (!token) { - this.observedUserLookups.delete(userId); - return undefined; - } + private hydrateObservedNames(): void { + if (this.hydratedObservedNames) return; + this.hydratedObservedNames = true; + const graph = this.persistedObservedContacts(); + if (!graph) return; + const hydrate = ( + cache: Map, + id: string, + label: string, + ): void => { + if (label !== id) cache.set(id, label); + }; + for (const user of graph.users) { + hydrate(this.observedUserNames, user.id, user.label); + } + for (const group of graph.groups) { + hydrate(this.observedChatNames, group.id, group.label); + for (const member of group.users) { + hydrate(this.observedUserNames, member.id, member.label); + } + } + } - const userIdType = userId.startsWith('ou_') - ? 'open_id' - : userId.startsWith('on_') - ? 'union_id' - : 'user_id'; - const response = await fetch( + private observedUserName(userId: string): Promise { + const userIdType = userId.startsWith('ou_') + ? 'open_id' + : userId.startsWith('on_') + ? 'union_id' + : 'user_id'; + return this.observedNameLookup({ + lookups: this.observedUserLookups, + names: this.observedUserNames, + id: userId, + request: (token) => + fetch( `${BASE_URL}/contact/v3/users/basic_batch?user_id_type=${userIdType}`, { method: 'POST', @@ -714,72 +737,91 @@ export class FeishuChannel extends ChannelBase { body: JSON.stringify({ user_ids: [userId] }), signal: AbortSignal.timeout(15_000), }, - ); - if (!response.ok) { - if (response.status === 401) this.tokenCache = undefined; - return undefined; - } - - const body = (await response.json()) as { + ), + extractName: (body) => { + const data = body as { code?: number; data?: { users?: Array<{ name?: string }> }; }; - const name = body.data?.users?.[0]?.name?.trim(); - if (body.code !== 0 || !name) return undefined; - this.observedUserNames.set(userId, name); - return name; - } catch { - return undefined; - } - })(); - this.observedUserLookups.set(userId, lookup); - return lookup; + return data.code === 0 ? data.data?.users?.[0]?.name : undefined; + }, + }); } private observedChatName(chatId: string): Promise { - const existing = this.observedChatLookups.get(chatId); + return this.observedNameLookup({ + lookups: this.observedChatLookups, + names: this.observedChatNames, + id: chatId, + request: (token) => + fetch(`${BASE_URL}/im/v1/chats/${encodeURIComponent(chatId)}`, { + headers: { Authorization: `Bearer ${token}` }, + signal: AbortSignal.timeout(15_000), + }), + extractName: (body) => { + const data = body as { code?: number; data?: { name?: string } }; + return data.code === 0 ? data.data?.name : undefined; + }, + }); + } + + private observedNameLookup(options: { + lookups: Map>; + names: Map; + id: string; + request: (token: string) => Promise; + extractName: (body: unknown) => string | undefined; + }): Promise { + const existing = options.lookups.get(options.id); if (existing) return existing; const lookup = (async () => { try { const token = await this.getTenantAccessToken(); if (!token) { - this.observedChatLookups.delete(chatId); + options.lookups.delete(options.id); return undefined; } - const response = await fetch( - `${BASE_URL}/im/v1/chats/${encodeURIComponent(chatId)}`, - { - headers: { Authorization: `Bearer ${token}` }, - signal: AbortSignal.timeout(15_000), - }, - ); + const response = await options.request(token); if (!response.ok) { - if (response.status === 401) this.tokenCache = undefined; + if (response.status === 401) { + this.tokenCache = undefined; + options.lookups.delete(options.id); + } return undefined; } - const body = (await response.json()) as { - code?: number; - data?: { name?: string }; - }; - const name = body.data?.name?.trim(); - if (body.code !== 0 || !name) return undefined; - this.observedChatNames.set(chatId, name); - return name; + const name = options.extractName(await response.json())?.trim(); + if (!name) return undefined; + const label = sanitizeSenderName(name); + if (label === 'unknown') return undefined; + options.names.set(options.id, label); + return label; } catch { return undefined; } })(); - this.observedChatLookups.set(chatId, lookup); + options.lookups.set(options.id, lookup); return lookup; } protected override onObservedContact(envelope: Envelope): void { + this.observedContactWrites.set(this.observedContactKey(envelope), { + senderName: envelope.senderName, + chatName: envelope.chatName, + }); void this.enrichObservedContact(envelope).catch(() => {}); } + private observedContactKey(envelope: Envelope): string { + return envelope.isGroup + ? `${envelope.senderId}\u0000${envelope.chatId}\u0000${ + envelope.threadId ?? '' + }` + : envelope.senderId; + } + private async enrichObservedContact(envelope: Envelope): Promise { const [senderName, chatName] = await Promise.all([ this.observedUserName(envelope.senderId), @@ -788,13 +830,20 @@ export class FeishuChannel extends ChannelBase { : Promise.resolve(undefined), ]); if (!senderName && !chatName) return; + const key = this.observedContactKey(envelope); + const nextLabels = { + senderName: senderName ?? envelope.senderName, + chatName: chatName ?? envelope.chatName, + }; + const persistedLabels = this.observedContactWrites.get(key); if ( - (!senderName || envelope.senderName === senderName) && - (!chatName || envelope.chatName === chatName) + persistedLabels && + persistedLabels.senderName === nextLabels.senderName && + persistedLabels.chatName === nextLabels.chatName ) { return; } - + this.observedContactWrites.set(key, nextLabels); await this.recordObservedContact({ ...envelope, ...(senderName ? { senderName } : {}), @@ -2077,6 +2126,7 @@ export class FeishuChannel extends ChannelBase { sender.sender_id?.user_id || sender.sender_id?.union_id || ''; + this.hydrateObservedNames(); const senderName = this.observedUserNames.get(senderId) || senderId; const chatName = isGroup ? this.observedChatNames.get(chatId) : undefined; diff --git a/packages/channels/feishu/src/adapter.test.ts b/packages/channels/feishu/src/adapter.test.ts index aa9d58dd599..cdde0fb787d 100644 --- a/packages/channels/feishu/src/adapter.test.ts +++ b/packages/channels/feishu/src/adapter.test.ts @@ -24,6 +24,7 @@ import type { ChannelConfig, ChannelProactiveDeliveryError, ChannelTaskLifecycleEvent, + ObservedChannelContactGraph, SessionTarget, } from '@qwen-code/channel-base'; @@ -88,6 +89,7 @@ class ObservedContactFeishuChannel extends FeishuChannel { function createObservedContactChannel( observe: NonNullable['observe'], + list?: NonNullable['list'], ): { channel: ObservedContactFeishuChannel; bridge: ChannelAgentBridge; @@ -101,7 +103,7 @@ function createObservedContactChannel( groups: { '*': { requireMention: false } }, }), bridge, - { observedContacts: { observe } }, + { observedContacts: { observe, ...(list ? { list } : {}) } }, ); Object.assign(channel as unknown as Record, { tokenCache: { @@ -133,6 +135,27 @@ function feishuGroupMessage(messageId: string): Record { }; } +function feishuDmMessage(messageId: string): Record { + return { + message: { + message_id: messageId, + chat_id: 'oc_dm', + chat_type: 'p2p', + message_type: 'text', + content: JSON.stringify({ text: 'hello' }), + }, + sender: { + sender_id: { + union_id: 'on_user', + user_id: 'user_1', + open_id: 'ou_user', + }, + sender_type: 'user', + tenant_key: 'tenant_1', + }, + }; +} + // Access private methods for unit testing function getPrivateMethod(instance: unknown, method: string): T { return (instance as Record)[method] as T; @@ -609,11 +632,460 @@ describe('FeishuChannel', () => { try { await expect(observedUserName('ou_first')).resolves.toBeUndefined(); await expect(observedUserName('ou_second')).resolves.toBe('Bob'); + await expect(observedUserName('ou_first')).resolves.toBe('Bob'); + expect(fetchSpy).toHaveBeenCalledTimes(4); + } finally { + fetchSpy.mockRestore(); + } + }); + + it('retries a chat lookup when token acquisition fails before the request', async () => { + const observe = vi.fn(); + const { channel } = createObservedContactChannel(observe); + Object.assign(channel as unknown as Record, { + tokenCache: undefined, + }); + const fetchSpy = vi + .spyOn(global, 'fetch') + .mockResolvedValueOnce(jsonResponse('unavailable', 503)) + .mockResolvedValueOnce( + jsonResponse({ + tenant_access_token: 'fresh_token', + expire: 3600, + }), + ) + .mockResolvedValueOnce( + jsonResponse({ code: 0, data: { name: 'Project Group' } }), + ); + const stderrSpy = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + const observedChatName = getPrivateMethod< + (chatId: string) => Promise + >(channel, 'observedChatName').bind(channel); + + try { + await expect(observedChatName('oc_group')).resolves.toBeUndefined(); + await expect(observedChatName('oc_group')).resolves.toBe( + 'Project Group', + ); + expect(fetchSpy).toHaveBeenCalledTimes(3); + } finally { + fetchSpy.mockRestore(); + stderrSpy.mockRestore(); + } + }); + + it('retries a 401 lookup for the same ID once the token refreshes', async () => { + const observe = vi.fn(); + const { channel } = createObservedContactChannel(observe); + const fetchSpy = vi + .spyOn(global, 'fetch') + .mockImplementation(async (input, init) => { + const url = String(input); + if (url.includes('/tenant_access_token/internal')) { + return jsonResponse({ + tenant_access_token: 'fresh_token', + expire: 3600, + }); + } + if (url.includes('/contact/v3/users/basic_batch')) { + const authorization = new Headers(init?.headers).get( + 'Authorization', + ); + if (authorization === 'Bearer test_token') { + return jsonResponse('unauthorized', 401); + } + return jsonResponse({ + code: 0, + data: { users: [{ name: 'Bob' }] }, + }); + } + throw new Error(`Unexpected request: ${url}`); + }); + const observedUserName = getPrivateMethod< + (userId: string) => Promise + >(channel, 'observedUserName').bind(channel); + + try { + await expect(observedUserName('ou_first')).resolves.toBeUndefined(); + await expect(observedUserName('ou_first')).resolves.toBe('Bob'); expect(fetchSpy).toHaveBeenCalledTimes(3); } finally { fetchSpy.mockRestore(); } }); + + it('sanitizes resolved display names before caching them', async () => { + const observe = vi.fn(); + const { channel } = createObservedContactChannel(observe); + const fetchSpy = vi + .spyOn(global, 'fetch') + .mockImplementation(async (input) => { + const url = String(input); + if (url.includes('/contact/v3/users/basic_batch')) { + return jsonResponse({ + code: 0, + data: { users: [{ name: 'Evil\rName' }] }, + }); + } + throw new Error(`Unexpected request: ${url}`); + }); + const observedUserName = getPrivateMethod< + (userId: string) => Promise + >(channel, 'observedUserName').bind(channel); + + try { + await expect(observedUserName('ou_user')).resolves.toBe('Evil Name'); + expect( + (channel as unknown as Record)['observedUserNames'], + ).toEqual(new Map([['ou_user', 'Evil Name']])); + } finally { + fetchSpy.mockRestore(); + } + }); + + it('writes an enriched observation when only the user lookup resolves', async () => { + const observe = vi.fn(); + const { channel } = createObservedContactChannel(observe); + const fetchSpy = vi + .spyOn(global, 'fetch') + .mockImplementation(async (input) => { + const url = String(input); + if (url.includes('/contact/v3/users/basic_batch')) { + return jsonResponse({ + code: 0, + data: { users: [{ name: 'Alice' }] }, + }); + } + if (url.includes('/im/v1/chats/oc_group')) { + return jsonResponse({ code: 99991672, msg: 'no permission' }); + } + throw new Error(`Unexpected request: ${url}`); + }); + const onMessage = getPrivateMethod<(data: unknown) => void>( + channel, + 'onMessage', + ).bind(channel); + + try { + onMessage(feishuGroupMessage('message_1')); + + await vi.waitFor(() => expect(observe).toHaveBeenCalledTimes(2)); + expect(observe).toHaveBeenNthCalledWith(1, 'test', { + user: { id: 'ou_user', label: 'ou_user' }, + group: { id: 'oc_group', label: 'oc_group' }, + }); + expect(observe).toHaveBeenNthCalledWith(2, 'test', { + user: { id: 'ou_user', label: 'Alice' }, + group: { id: 'oc_group', label: 'oc_group' }, + }); + + onMessage(feishuGroupMessage('message_2')); + + await vi.waitFor(() => expect(observe).toHaveBeenCalledTimes(3)); + expect(observe).toHaveBeenNthCalledWith(3, 'test', { + user: { id: 'ou_user', label: 'Alice' }, + group: { id: 'oc_group', label: 'oc_group' }, + }); + expect(fetchSpy).toHaveBeenCalledTimes(2); + } finally { + fetchSpy.mockRestore(); + } + }); + + it('enriches direct-chat senders without issuing chat lookups', async () => { + const observe = vi.fn(); + const { channel } = createObservedContactChannel(observe); + const fetchSpy = vi + .spyOn(global, 'fetch') + .mockImplementation(async (input) => { + const url = String(input); + if (url.includes('/contact/v3/users/basic_batch')) { + return jsonResponse({ + code: 0, + data: { users: [{ name: 'Alice' }] }, + }); + } + throw new Error(`Unexpected request: ${url}`); + }); + const onMessage = getPrivateMethod<(data: unknown) => void>( + channel, + 'onMessage', + ).bind(channel); + + try { + onMessage(feishuDmMessage('dm_1')); + + await vi.waitFor(() => expect(observe).toHaveBeenCalledTimes(2)); + expect(observe).toHaveBeenNthCalledWith(1, 'test', { + user: { id: 'ou_user', label: 'ou_user' }, + }); + expect(observe).toHaveBeenNthCalledWith(2, 'test', { + user: { id: 'ou_user', label: 'Alice' }, + }); + + onMessage(feishuDmMessage('dm_2')); + + await vi.waitFor(() => expect(observe).toHaveBeenCalledTimes(3)); + expect(observe).toHaveBeenNthCalledWith(3, 'test', { + user: { id: 'ou_user', label: 'Alice' }, + }); + expect(fetchSpy).toHaveBeenCalledTimes(1); + expect( + fetchSpy.mock.calls.some(([input]) => + String(input).includes('/im/v1/chats/'), + ), + ).toBe(false); + } finally { + fetchSpy.mockRestore(); + } + }); + + it('maps non-open_id senders to the matching user_id_type', async () => { + const observe = vi.fn(); + const { channel } = createObservedContactChannel(observe); + const fetchSpy = vi + .spyOn(global, 'fetch') + .mockImplementation(async (input) => { + const url = String(input); + if (url.includes('/contact/v3/users/basic_batch')) { + return jsonResponse({ + code: 0, + data: { users: [{ name: 'Named' }] }, + }); + } + if (url.includes('/im/v1/chats/oc_group')) { + return jsonResponse({ + code: 0, + data: { name: 'Project Group' }, + }); + } + throw new Error(`Unexpected request: ${url}`); + }); + const onMessage = getPrivateMethod<(data: unknown) => void>( + channel, + 'onMessage', + ).bind(channel); + + try { + const unionOnly = feishuGroupMessage('message_union'); + (unionOnly['sender'] as Record)['sender_id'] = { + union_id: 'on_user', + }; + onMessage(unionOnly); + await vi.waitFor(() => + expect(fetchSpy).toHaveBeenCalledWith( + expect.stringContaining('user_id_type=union_id'), + expect.objectContaining({ + body: JSON.stringify({ user_ids: ['on_user'] }), + }), + ), + ); + + const userIdOnly = feishuGroupMessage('message_user'); + (userIdOnly['sender'] as Record)['sender_id'] = { + user_id: 'user_1', + }; + onMessage(userIdOnly); + await vi.waitFor(() => + expect(fetchSpy).toHaveBeenCalledWith( + expect.stringContaining('user_id_type=user_id'), + expect.objectContaining({ + body: JSON.stringify({ user_ids: ['user_1'] }), + }), + ), + ); + expect( + fetchSpy.mock.calls.some(([input]) => + String(input).includes('user_id_type=open_id'), + ), + ).toBe(false); + } finally { + fetchSpy.mockRestore(); + } + }); + + it('shares one in-flight lookup across concurrent messages', async () => { + const observe = vi.fn(); + const { channel } = createObservedContactChannel(observe); + const userLookup = deferred(); + const chatLookup = deferred(); + const fetchSpy = vi.spyOn(global, 'fetch').mockImplementation((input) => { + const url = String(input); + if (url.includes('/contact/v3/users/basic_batch')) { + return userLookup.promise; + } + if (url.includes('/im/v1/chats/oc_group')) { + return chatLookup.promise; + } + throw new Error(`Unexpected request: ${url}`); + }); + const onMessage = getPrivateMethod<(data: unknown) => void>( + channel, + 'onMessage', + ).bind(channel); + + try { + onMessage(feishuGroupMessage('message_1')); + onMessage(feishuGroupMessage('message_2')); + await vi.waitFor(() => { + expect(observe).toHaveBeenCalledTimes(2); + expect(fetchSpy).toHaveBeenCalledTimes(2); + }); + + userLookup.resolve( + jsonResponse({ code: 0, data: { users: [{ name: 'Alice' }] } }), + ); + chatLookup.resolve( + jsonResponse({ code: 0, data: { name: 'Project Group' } }), + ); + + await vi.waitFor(() => expect(observe).toHaveBeenCalledTimes(3)); + expect(observe).toHaveBeenLastCalledWith('test', { + user: { id: 'ou_user', label: 'Alice' }, + group: { id: 'oc_group', label: 'Project Group' }, + }); + expect( + fetchSpy.mock.calls.filter(([input]) => + String(input).includes('/contact/v3/users/basic_batch'), + ), + ).toHaveLength(1); + expect( + fetchSpy.mock.calls.filter(([input]) => + String(input).includes('/im/v1/chats/oc_group'), + ), + ).toHaveLength(1); + } finally { + fetchSpy.mockRestore(); + } + }); + + it('hydrates label caches from persisted observations after a restart', async () => { + const observe = vi.fn(); + const list = vi.fn( + (): ObservedChannelContactGraph => ({ + users: [], + groups: [ + { + channelName: 'test', + id: 'oc_group', + label: 'Project Group', + lastObservedAt: '2026-01-01T00:00:00.000Z', + users: [ + { + id: 'ou_user', + label: 'Alice', + lastObservedAt: '2026-01-01T00:00:00.000Z', + }, + ], + topics: [], + }, + { + channelName: 'other', + id: 'oc_foreign', + label: 'Foreign Group', + lastObservedAt: '2026-01-01T00:00:00.000Z', + users: [ + { + id: 'ou_foreign', + label: 'Foreign User', + lastObservedAt: '2026-01-01T00:00:00.000Z', + }, + ], + topics: [], + }, + ], + }), + ); + const { channel, bridge } = createObservedContactChannel(observe, list); + const fetchSpy = vi + .spyOn(global, 'fetch') + .mockResolvedValue(jsonResponse('rate limited', 429)); + const onMessage = getPrivateMethod<(data: unknown) => void>( + channel, + 'onMessage', + ).bind(channel); + + try { + onMessage(feishuGroupMessage('message_1')); + + await vi.waitFor(() => { + expect(observe).toHaveBeenCalledTimes(1); + expect(bridge.prompt).toHaveBeenCalledTimes(1); + }); + expect(observe).toHaveBeenNthCalledWith(1, 'test', { + user: { id: 'ou_user', label: 'Alice' }, + group: { id: 'oc_group', label: 'Project Group' }, + }); + + onMessage(feishuGroupMessage('message_2')); + + await vi.waitFor(() => expect(observe).toHaveBeenCalledTimes(2)); + expect(observe).toHaveBeenNthCalledWith(2, 'test', { + user: { id: 'ou_user', label: 'Alice' }, + group: { id: 'oc_group', label: 'Project Group' }, + }); + expect(list).toHaveBeenCalledTimes(1); + expect(fetchSpy).toHaveBeenCalledTimes(2); + } finally { + fetchSpy.mockRestore(); + } + }); + + it('attaches cached labels to later envelopes and prompts', async () => { + const observe = vi.fn(); + const { channel, bridge } = createObservedContactChannel(observe); + const fetchSpy = vi + .spyOn(global, 'fetch') + .mockImplementation(async (input) => { + const url = String(input); + if (url.includes('/contact/v3/users/basic_batch')) { + return jsonResponse({ + code: 0, + data: { users: [{ name: 'Alice' }] }, + }); + } + if (url.includes('/im/v1/chats/oc_group')) { + return jsonResponse({ + code: 0, + data: { name: 'Project Group' }, + }); + } + throw new Error(`Unexpected request: ${url}`); + }); + const onMessage = getPrivateMethod<(data: unknown) => void>( + channel, + 'onMessage', + ).bind(channel); + + try { + onMessage(feishuGroupMessage('message_1')); + await vi.waitFor(() => expect(observe).toHaveBeenCalledTimes(2)); + + onMessage(feishuGroupMessage('message_2')); + await vi.waitFor(() => expect(bridge.prompt).toHaveBeenCalledTimes(2)); + + expect(bridge.prompt).toHaveBeenNthCalledWith( + 1, + expect.any(String), + expect.stringContaining('[ou_user]'), + expect.anything(), + ); + expect(bridge.prompt).toHaveBeenNthCalledWith( + 2, + expect.any(String), + expect.stringContaining('[Alice]'), + expect.anything(), + ); + expect(observe).toHaveBeenNthCalledWith(3, 'test', { + user: { id: 'ou_user', label: 'Alice' }, + group: { id: 'oc_group', label: 'Project Group' }, + }); + } finally { + fetchSpy.mockRestore(); + } + }); }); describe('extractCardText', () => { diff --git a/packages/cli/src/commands/channel/daemon-worker.test.ts b/packages/cli/src/commands/channel/daemon-worker.test.ts index fc903cdb2ea..3fa19d874bf 100644 --- a/packages/cli/src/commands/channel/daemon-worker.test.ts +++ b/packages/cli/src/commands/channel/daemon-worker.test.ts @@ -40,9 +40,11 @@ const mockDaemonChannelStateDir = vi.hoisted(() => ), ); const mockObserveContact = vi.hoisted(() => vi.fn()); +const mockListContacts = vi.hoisted(() => vi.fn()); const mockObservedContactStore = vi.hoisted(() => vi.fn(() => ({ observe: mockObserveContact, + list: mockListContacts, })), ); const mockLoadSettings = vi.hoisted(() => @@ -246,6 +248,7 @@ vi.mock('./runtime.js', () => ({ })); vi.mock('./observed-contact-store.js', () => ({ + OBSERVED_CONTACT_MAX_FRESH_WITHIN_SECONDS: 365 * 24 * 60 * 60, ObservedChannelContactStore: mockObservedContactStore, })); @@ -798,6 +801,7 @@ describe('runChannelDaemonWorker', () => { channelMemoryRecallObserver: mockRecordChannelMemoryRecallMetrics, observedContacts: { observe: expect.any(Function), + list: expect.any(Function), }, stateDir: '/tmp/qwen/channels/daemon/workspace-hash/instances/telegram-hash', @@ -814,6 +818,7 @@ describe('runChannelDaemonWorker', () => { const channelOptions = mockCreateChannel.mock.calls[0]![3] as { observedContacts: { observe(channelName: string, observation: unknown): unknown; + list(): unknown; }; }; const observation = { @@ -822,6 +827,10 @@ describe('runChannelDaemonWorker', () => { }; channelOptions.observedContacts.observe('telegram', observation); expect(mockObserveContact).toHaveBeenCalledWith('telegram', observation); + channelOptions.observedContacts.list(); + expect(mockListContacts).toHaveBeenCalledWith({ + freshWithinSeconds: 365 * 24 * 60 * 60, + }); expect(mockRegisterPermissionRelay).toHaveBeenCalledWith( bridgeFacade, mockSessionRouter.mock.results[0]!.value, diff --git a/packages/cli/src/commands/channel/daemon-worker.ts b/packages/cli/src/commands/channel/daemon-worker.ts index 6067633a23d..3aa6d3fa953 100644 --- a/packages/cli/src/commands/channel/daemon-worker.ts +++ b/packages/cli/src/commands/channel/daemon-worker.ts @@ -84,7 +84,10 @@ import { type ParsedChannel, } from './runtime.js'; import { BridgeChannelMemoryIntentClassifier } from './memory-intent-classifier.js'; -import { ObservedChannelContactStore } from './observed-contact-store.js'; +import { + OBSERVED_CONTACT_MAX_FRESH_WITHIN_SECONDS, + ObservedChannelContactStore, +} from './observed-contact-store.js'; import { createChannelLoopController, isChannelCronEnabled, @@ -554,6 +557,10 @@ export async function runChannelDaemonWorker( observe: (channelName, observation) => { observedContacts.observe(channelName, observation); }, + list: () => + observedContacts.list({ + freshWithinSeconds: OBSERVED_CONTACT_MAX_FRESH_WITHIN_SECONDS, + }), }, ...(loopController ? { loopController } : {}), }), From 93bd433dacd1d7244f212331150dc8ae53c4416c Mon Sep 17 00:00:00 2001 From: hit_aran Date: Fri, 7 Aug 2026 12:11:59 +0800 Subject: [PATCH 6/7] fix(feishu): bound label caches, honor observation recency, silence enrichment token failures (#8569) - hydrate runtime label caches from the newest observation per contact so stale group membership labels cannot overwrite more recent ones - cap the user/chat label, in-flight lookup, and write-dedup maps at 500 entries (matching the persisted registry) and evict oldest entries - route best-effort label lookups through a silent token refresh path so enrichment failures no longer write to stderr - add tests for silent token refresh, newest-label hydration, cache cap, and the persisted-observation reject path in hook ordering --- .../channels/base/src/ChannelBase.test.ts | 36 +++++ packages/channels/feishu/src/FeishuAdapter.ts | 78 ++++++++--- packages/channels/feishu/src/adapter.test.ts | 123 ++++++++++++++++++ 3 files changed, 221 insertions(+), 16 deletions(-) diff --git a/packages/channels/base/src/ChannelBase.test.ts b/packages/channels/base/src/ChannelBase.test.ts index cea902c5a57..1e8e7afdcd3 100644 --- a/packages/channels/base/src/ChannelBase.test.ts +++ b/packages/channels/base/src/ChannelBase.test.ts @@ -933,6 +933,42 @@ describe('ChannelBase', () => { expect(bridge.prompt).toHaveBeenCalled(); }); + it('still notifies the adapter after a rejected contact persistence', async () => { + const order: string[] = []; + class ObservedHookChannel extends TestChannel { + readonly observedEnvelopes: Envelope[] = []; + + protected override onObservedContact(envelope: Envelope): void { + order.push('hook'); + this.observedEnvelopes.push(envelope); + } + } + + const observe = vi.fn(async () => { + throw new Error('persistence unavailable'); + }); + const stderrSpy = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + const ch = new ObservedHookChannel('test-chan', defaultConfig(), bridge, { + observedContacts: { observe }, + }); + const message = envelope(); + + await ch.handleInbound(message); + + const stderrOutput = stderrSpy.mock.calls + .map((call) => String(call[0])) + .join(''); + stderrSpy.mockRestore(); + + expect(observe).toHaveBeenCalledTimes(1); + expect(order).toEqual(['hook']); + expect(ch.observedEnvelopes).toEqual([message]); + expect(bridge.prompt).toHaveBeenCalled(); + expect(stderrOutput).toContain('observed contact persistence failed'); + }); + it('falls back to the complete sender ID for an unusable label', async () => { const observe = vi.fn(); const ch = createChannel({}, { observedContacts: { observe } }); diff --git a/packages/channels/feishu/src/FeishuAdapter.ts b/packages/channels/feishu/src/FeishuAdapter.ts index 0ae90855e32..c0f1aed2e8b 100644 --- a/packages/channels/feishu/src/FeishuAdapter.ts +++ b/packages/channels/feishu/src/FeishuAdapter.ts @@ -84,6 +84,12 @@ interface CardSessionState { /** Track seen message IDs to deduplicate retried events. */ const DEDUP_TTL_MS = 5 * 60 * 1000; +/** + * Runtime label/lookup caches are bounded like the persisted observed-contact + * registry (500 observations) so a long-running daemon does not retain every + * user/chat/thread ID it ever sees. + */ +const OBSERVED_LABEL_CACHE_LIMIT = 500; /** Minimum interval between card updates (ms) to avoid API rate limiting. */ const CARD_UPDATE_INTERVAL_MS = 1500; @@ -637,13 +643,15 @@ export class FeishuChannel extends ChannelBase { return text.trim() || undefined; } - private async getTenantAccessToken(): Promise { + private async getTenantAccessToken(options?: { + silent?: boolean; + }): Promise { if (this.tokenCache && Date.now() < this.tokenCache.expiresAt) { return this.tokenCache.token; } if (this.tokenRefreshPromise) return this.tokenRefreshPromise; - this.tokenRefreshPromise = this.refreshToken(); + this.tokenRefreshPromise = this.refreshToken(options); try { return await this.tokenRefreshPromise; } finally { @@ -651,7 +659,12 @@ export class FeishuChannel extends ChannelBase { } } - private async refreshToken(): Promise { + private async refreshToken(options?: { + silent?: boolean; + }): Promise { + // Best-effort label enrichment must stay silent on failure; core delivery + // paths keep logging token errors. + const silent = options?.silent === true; try { const resp = await fetch( `${BASE_URL}/auth/v3/tenant_access_token/internal`, @@ -667,9 +680,11 @@ export class FeishuChannel extends ChannelBase { ); if (!resp.ok) { - process.stderr.write( - `[Feishu:${this.name}] getTenantAccessToken failed: HTTP ${resp.status}\n`, - ); + if (!silent) { + process.stderr.write( + `[Feishu:${this.name}] getTenantAccessToken failed: HTTP ${resp.status}\n`, + ); + } if (resp.status === 401) this.tokenCache = undefined; return undefined; } @@ -685,9 +700,11 @@ export class FeishuChannel extends ChannelBase { }; return this.tokenCache.token; } catch (err) { - process.stderr.write( - `[Feishu:${this.name}] getTenantAccessToken error: ${err}\n`, - ); + if (!silent) { + process.stderr.write( + `[Feishu:${this.name}] getTenantAccessToken error: ${err}\n`, + ); + } return undefined; } } @@ -697,22 +714,47 @@ export class FeishuChannel extends ChannelBase { this.hydratedObservedNames = true; const graph = this.persistedObservedContacts(); if (!graph) return; - const hydrate = ( - cache: Map, + // Select the newest non-ID label per contact so an older observation + // (for example a stale group membership) cannot overwrite a more recent + // one during the traversal. + const newestUser = new Map(); + const newestChat = new Map(); + const consider = ( + best: Map, id: string, label: string, + at: string, ): void => { - if (label !== id) cache.set(id, label); + if (label === id) return; + const current = best.get(id); + if (!current || at >= current.at) best.set(id, { label, at }); }; for (const user of graph.users) { - hydrate(this.observedUserNames, user.id, user.label); + consider(newestUser, user.id, user.label, user.lastObservedAt); } for (const group of graph.groups) { - hydrate(this.observedChatNames, group.id, group.label); + consider(newestChat, group.id, group.label, group.lastObservedAt); for (const member of group.users) { - hydrate(this.observedUserNames, member.id, member.label); + consider(newestUser, member.id, member.label, member.lastObservedAt); } } + for (const [id, entry] of newestUser) { + this.observedUserNames.set(id, entry.label); + } + for (const [id, entry] of newestChat) { + this.observedChatNames.set(id, entry.label); + } + this.capObservedCache(this.observedUserNames); + this.capObservedCache(this.observedChatNames); + } + + /** Evicts the oldest-inserted entries once a runtime cache exceeds the cap. */ + private capObservedCache(cache: Map): void { + while (cache.size > OBSERVED_LABEL_CACHE_LIMIT) { + const oldest = cache.keys().next(); + if (oldest.done) break; + cache.delete(oldest.value); + } } private observedUserName(userId: string): Promise { @@ -777,7 +819,7 @@ export class FeishuChannel extends ChannelBase { const lookup = (async () => { try { - const token = await this.getTenantAccessToken(); + const token = await this.getTenantAccessToken({ silent: true }); if (!token) { options.lookups.delete(options.id); return undefined; @@ -797,12 +839,14 @@ export class FeishuChannel extends ChannelBase { const label = sanitizeSenderName(name); if (label === 'unknown') return undefined; options.names.set(options.id, label); + this.capObservedCache(options.names); return label; } catch { return undefined; } })(); options.lookups.set(options.id, lookup); + this.capObservedCache(options.lookups); return lookup; } @@ -811,6 +855,7 @@ export class FeishuChannel extends ChannelBase { senderName: envelope.senderName, chatName: envelope.chatName, }); + this.capObservedCache(this.observedContactWrites); void this.enrichObservedContact(envelope).catch(() => {}); } @@ -844,6 +889,7 @@ export class FeishuChannel extends ChannelBase { return; } this.observedContactWrites.set(key, nextLabels); + this.capObservedCache(this.observedContactWrites); await this.recordObservedContact({ ...envelope, ...(senderName ? { senderName } : {}), diff --git a/packages/channels/feishu/src/adapter.test.ts b/packages/channels/feishu/src/adapter.test.ts index cdde0fb787d..38f947e082a 100644 --- a/packages/channels/feishu/src/adapter.test.ts +++ b/packages/channels/feishu/src/adapter.test.ts @@ -1086,6 +1086,129 @@ describe('FeishuChannel', () => { fetchSpy.mockRestore(); } }); + + it('keeps enrichment silent when tenant token acquisition fails', async () => { + const observe = vi.fn(); + const { channel, bridge } = createObservedContactChannel(observe); + Object.assign(channel as unknown as Record, { + tokenCache: undefined, + }); + const fetchSpy = vi + .spyOn(global, 'fetch') + .mockImplementation(async (input) => { + const url = String(input); + if (url.includes('/auth/v3/tenant_access_token/internal')) { + throw new Error('token endpoint unavailable'); + } + throw new Error(`Unexpected request: ${url}`); + }); + const stderrSpy = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + const onMessage = getPrivateMethod<(data: unknown) => void>( + channel, + 'onMessage', + ).bind(channel); + + try { + onMessage(feishuGroupMessage('message_1')); + + await vi.waitFor(() => { + expect(observe).toHaveBeenCalledTimes(1); + expect(bridge.prompt).toHaveBeenCalledTimes(1); + }); + + expect(observe).toHaveBeenNthCalledWith(1, 'test', { + user: { id: 'ou_user', label: 'ou_user' }, + group: { id: 'oc_group', label: 'oc_group' }, + }); + expect(stderrSpy).not.toHaveBeenCalled(); + } finally { + fetchSpy.mockRestore(); + stderrSpy.mockRestore(); + } + }); + + it('prefers the newest persisted label when hydrating overlapping contacts', async () => { + const observe = vi.fn(); + const list = vi.fn( + (): ObservedChannelContactGraph => ({ + users: [ + { + channelName: 'test', + id: 'ou_user', + label: 'New Name', + lastObservedAt: '2026-02-01T00:00:00.000Z', + }, + ], + groups: [ + { + channelName: 'test', + id: 'oc_group', + label: 'Project Group', + lastObservedAt: '2026-01-01T00:00:00.000Z', + users: [ + { + id: 'ou_user', + label: 'Old Name', + lastObservedAt: '2026-01-01T00:00:00.000Z', + }, + ], + topics: [], + }, + ], + }), + ); + const { channel, bridge } = createObservedContactChannel(observe, list); + const fetchSpy = vi + .spyOn(global, 'fetch') + .mockResolvedValue(jsonResponse('rate limited', 429)); + const onMessage = getPrivateMethod<(data: unknown) => void>( + channel, + 'onMessage', + ).bind(channel); + + try { + onMessage(feishuGroupMessage('message_1')); + + await vi.waitFor(() => { + expect(observe).toHaveBeenCalledTimes(1); + expect(bridge.prompt).toHaveBeenCalledTimes(1); + }); + + expect(bridge.prompt).toHaveBeenCalledWith( + expect.any(String), + expect.stringContaining('[New Name]'), + expect.anything(), + ); + expect(observe).toHaveBeenNthCalledWith(1, 'test', { + user: { id: 'ou_user', label: 'New Name' }, + group: { id: 'oc_group', label: 'Project Group' }, + }); + } finally { + fetchSpy.mockRestore(); + } + }); + + it('bounds runtime label caches to the observed-contact retention cap', () => { + const observe = vi.fn(); + const { channel } = createObservedContactChannel(observe); + const capObservedCache = getPrivateMethod< + (cache: Map) => void + >(channel, 'capObservedCache').bind(channel); + const cache = new Map(); + for (let i = 0; i < 525; i++) { + cache.set(`id_${i}`, `label_${i}`); + } + + capObservedCache(cache); + + expect(cache.size).toBe(500); + expect(cache.has('id_0')).toBe(false); + expect(cache.has('id_24')).toBe(false); + expect(cache.has('id_25')).toBe(true); + expect(cache.has('id_524')).toBe(true); + }); }); describe('extractCardText', () => { From 119e565317ad821368f64aac48877cb12d0d1464 Mon Sep 17 00:00:00 2001 From: qwen-code-ci-bot Date: Fri, 7 Aug 2026 11:20:19 +0000 Subject: [PATCH 7/7] fix(feishu): address observed-contact label review feedback (#8569) * Track core (non-silent) waiters on the shared tenant-token refresh so a silent-initiated refresh still logs token errors for joined delivery callers. * Short-circuit label lookups on the resolved names cache so evicted lookup entries do not trigger redundant API requests. * Re-hydrate label caches from the persisted registry after an in-lifetime cache eviction so the next initial write cannot clobber a persisted label with the raw ID. * Add mutation-proof regression tests for the channel-isolation filter, the list-failure swallow, the silent HTTP-error branch, and the 'unknown' label guard. --- packages/channels/feishu/src/FeishuAdapter.ts | 34 ++- packages/channels/feishu/src/adapter.test.ts | 240 ++++++++++++++++++ 2 files changed, 263 insertions(+), 11 deletions(-) diff --git a/packages/channels/feishu/src/FeishuAdapter.ts b/packages/channels/feishu/src/FeishuAdapter.ts index c0f1aed2e8b..f9ab0755c94 100644 --- a/packages/channels/feishu/src/FeishuAdapter.ts +++ b/packages/channels/feishu/src/FeishuAdapter.ts @@ -149,6 +149,9 @@ export class FeishuChannel extends ChannelBase { private botOpenId?: string; private tokenCache?: { token: string; expiresAt: number }; private tokenRefreshPromise?: Promise; + // Core (non-silent) callers waiting on the shared token refresh, so a + // silent-initiated refresh still logs token errors for them. + private tokenRefreshHasCoreWaiters = false; private readonly observedUserNames = new Map(); private readonly observedChatNames = new Map(); private readonly observedUserLookups = new Map< @@ -650,21 +653,20 @@ export class FeishuChannel extends ChannelBase { return this.tokenCache.token; } + if (!options?.silent) this.tokenRefreshHasCoreWaiters = true; if (this.tokenRefreshPromise) return this.tokenRefreshPromise; - this.tokenRefreshPromise = this.refreshToken(options); + this.tokenRefreshPromise = this.refreshToken(); try { return await this.tokenRefreshPromise; } finally { this.tokenRefreshPromise = undefined; + this.tokenRefreshHasCoreWaiters = false; } } - private async refreshToken(options?: { - silent?: boolean; - }): Promise { - // Best-effort label enrichment must stay silent on failure; core delivery - // paths keep logging token errors. - const silent = options?.silent === true; + private async refreshToken(): Promise { + // Best-effort label enrichment initiates silent refreshes; failures must + // still surface when a core delivery caller initiated or joined it. try { const resp = await fetch( `${BASE_URL}/auth/v3/tenant_access_token/internal`, @@ -680,7 +682,7 @@ export class FeishuChannel extends ChannelBase { ); if (!resp.ok) { - if (!silent) { + if (this.tokenRefreshHasCoreWaiters) { process.stderr.write( `[Feishu:${this.name}] getTenantAccessToken failed: HTTP ${resp.status}\n`, ); @@ -700,7 +702,7 @@ export class FeishuChannel extends ChannelBase { }; return this.tokenCache.token; } catch (err) { - if (!silent) { + if (this.tokenRefreshHasCoreWaiters) { process.stderr.write( `[Feishu:${this.name}] getTenantAccessToken error: ${err}\n`, ); @@ -749,12 +751,15 @@ export class FeishuChannel extends ChannelBase { } /** Evicts the oldest-inserted entries once a runtime cache exceeds the cap. */ - private capObservedCache(cache: Map): void { + private capObservedCache(cache: Map): boolean { + let evicted = false; while (cache.size > OBSERVED_LABEL_CACHE_LIMIT) { const oldest = cache.keys().next(); if (oldest.done) break; cache.delete(oldest.value); + evicted = true; } + return evicted; } private observedUserName(userId: string): Promise { @@ -814,6 +819,8 @@ export class FeishuChannel extends ChannelBase { request: (token: string) => Promise; extractName: (body: unknown) => string | undefined; }): Promise { + const cached = options.names.get(options.id); + if (cached) return Promise.resolve(cached); const existing = options.lookups.get(options.id); if (existing) return existing; @@ -839,7 +846,12 @@ export class FeishuChannel extends ChannelBase { const label = sanitizeSenderName(name); if (label === 'unknown') return undefined; options.names.set(options.id, label); - this.capObservedCache(options.names); + // Evicting a resolved label drops the next envelope back to the raw + // ID, and the initial persistence write would clobber the persisted + // label, so re-hydrate from the registry on the next message. + if (this.capObservedCache(options.names)) { + this.hydratedObservedNames = false; + } return label; } catch { return undefined; diff --git a/packages/channels/feishu/src/adapter.test.ts b/packages/channels/feishu/src/adapter.test.ts index 38f947e082a..dc159f66706 100644 --- a/packages/channels/feishu/src/adapter.test.ts +++ b/packages/channels/feishu/src/adapter.test.ts @@ -592,6 +592,7 @@ describe('FeishuChannel', () => { await expect(observedUserName('ou_user')).resolves.toBeUndefined(); await expect(observedUserName('ou_user')).resolves.toBe('Alice'); expect(fetchSpy).toHaveBeenCalledTimes(3); + expect(stderrSpy).not.toHaveBeenCalled(); } finally { fetchSpy.mockRestore(); stderrSpy.mockRestore(); @@ -670,6 +671,7 @@ describe('FeishuChannel', () => { 'Project Group', ); expect(fetchSpy).toHaveBeenCalledTimes(3); + expect(stderrSpy).not.toHaveBeenCalled(); } finally { fetchSpy.mockRestore(); stderrSpy.mockRestore(); @@ -745,6 +747,73 @@ describe('FeishuChannel', () => { } }); + it('reuses a cached label without a new lookup after lookup entries evict', async () => { + const observe = vi.fn(); + const { channel } = createObservedContactChannel(observe); + const fetchSpy = vi + .spyOn(global, 'fetch') + .mockImplementation(async (input) => { + const url = String(input); + if (url.includes('/contact/v3/users/basic_batch')) { + return jsonResponse({ + code: 0, + data: { users: [{ name: 'Alice' }] }, + }); + } + throw new Error(`Unexpected request: ${url}`); + }); + const observedUserName = getPrivateMethod< + (userId: string) => Promise + >(channel, 'observedUserName').bind(channel); + + try { + await expect(observedUserName('ou_user')).resolves.toBe('Alice'); + expect(fetchSpy).toHaveBeenCalledTimes(1); + + // Simulate FIFO eviction of the resolved lookup entry while the + // cached label survives. + ( + (channel as unknown as Record)[ + 'observedUserLookups' + ] as Map + ).clear(); + + await expect(observedUserName('ou_user')).resolves.toBe('Alice'); + expect(fetchSpy).toHaveBeenCalledTimes(1); + } finally { + fetchSpy.mockRestore(); + } + }); + + it('keeps the ID label when a resolved name sanitizes to unknown', async () => { + const observe = vi.fn(); + const { channel } = createObservedContactChannel(observe); + const fetchSpy = vi + .spyOn(global, 'fetch') + .mockImplementation(async (input) => { + const url = String(input); + if (url.includes('/contact/v3/users/basic_batch')) { + return jsonResponse({ + code: 0, + data: { users: [{ name: '\u200b\u200c\u200d' }] }, + }); + } + throw new Error(`Unexpected request: ${url}`); + }); + const observedUserName = getPrivateMethod< + (userId: string) => Promise + >(channel, 'observedUserName').bind(channel); + + try { + await expect(observedUserName('ou_user')).resolves.toBeUndefined(); + expect( + (channel as unknown as Record)['observedUserNames'], + ).toEqual(new Map()); + } finally { + fetchSpy.mockRestore(); + } + }); + it('writes an enriched observation when only the user lookup resolves', async () => { const observe = vi.fn(); const { channel } = createObservedContactChannel(observe); @@ -1027,12 +1096,76 @@ describe('FeishuChannel', () => { group: { id: 'oc_group', label: 'Project Group' }, }); expect(list).toHaveBeenCalledTimes(1); + // Hydrated labels short-circuit the enrichment lookups. + expect(fetchSpy).toHaveBeenCalledTimes(0); + + // Labels persisted by another channel instance must not hydrate into + // this channel's caches. + const foreign = feishuGroupMessage('message_3'); + (foreign['message'] as Record)['chat_id'] = + 'oc_foreign'; + (foreign['sender'] as Record)['sender_id'] = { + open_id: 'ou_foreign', + }; + onMessage(foreign); + + await vi.waitFor(() => { + expect(observe).toHaveBeenCalledTimes(3); + expect(bridge.prompt).toHaveBeenCalledTimes(3); + }); + expect(observe).toHaveBeenNthCalledWith(3, 'test', { + user: { id: 'ou_foreign', label: 'ou_foreign' }, + group: { id: 'oc_foreign', label: 'oc_foreign' }, + }); + expect(bridge.prompt).toHaveBeenNthCalledWith( + 3, + expect.any(String), + expect.stringContaining('[ou_foreign]'), + expect.anything(), + ); + expect(list).toHaveBeenCalledTimes(1); expect(fetchSpy).toHaveBeenCalledTimes(2); } finally { fetchSpy.mockRestore(); } }); + it('still processes messages when persisted observations cannot be listed', async () => { + const observe = vi.fn(); + const list = vi.fn((): ObservedChannelContactGraph => { + throw new Error('Invalid observed contact registry.'); + }); + const { channel, bridge } = createObservedContactChannel(observe, list); + const fetchSpy = vi + .spyOn(global, 'fetch') + .mockResolvedValue(jsonResponse('rate limited', 429)); + const stderrSpy = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + const onMessage = getPrivateMethod<(data: unknown) => void>( + channel, + 'onMessage', + ).bind(channel); + + try { + onMessage(feishuGroupMessage('message_1')); + + await vi.waitFor(() => { + expect(observe).toHaveBeenCalledTimes(1); + expect(bridge.prompt).toHaveBeenCalledTimes(1); + }); + expect(list).toHaveBeenCalledTimes(1); + expect(observe).toHaveBeenNthCalledWith(1, 'test', { + user: { id: 'ou_user', label: 'ou_user' }, + group: { id: 'oc_group', label: 'oc_group' }, + }); + expect(stderrSpy).not.toHaveBeenCalled(); + } finally { + fetchSpy.mockRestore(); + stderrSpy.mockRestore(); + } + }); + it('attaches cached labels to later envelopes and prompts', async () => { const observe = vi.fn(); const { channel, bridge } = createObservedContactChannel(observe); @@ -1129,6 +1262,38 @@ describe('FeishuChannel', () => { } }); + it('logs token errors for core callers that join a silent-initiated refresh', async () => { + const observe = vi.fn(); + const { channel } = createObservedContactChannel(observe); + Object.assign(channel as unknown as Record, { + tokenCache: undefined, + }); + const fetchSpy = vi + .spyOn(global, 'fetch') + .mockResolvedValue(jsonResponse('unavailable', 503)); + const stderrSpy = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + const getTenantAccessToken = getPrivateMethod< + (options?: { silent?: boolean }) => Promise + >(channel, 'getTenantAccessToken').bind(channel); + + try { + const results = await Promise.all([ + getTenantAccessToken({ silent: true }), + getTenantAccessToken(), + ]); + + expect(results).toEqual([undefined, undefined]); + expect(fetchSpy).toHaveBeenCalledTimes(1); + expect(stderrSpy).toHaveBeenCalledTimes(1); + expect(String(stderrSpy.mock.calls[0][0])).toContain('HTTP 503'); + } finally { + fetchSpy.mockRestore(); + stderrSpy.mockRestore(); + } + }); + it('prefers the newest persisted label when hydrating overlapping contacts', async () => { const observe = vi.fn(); const list = vi.fn( @@ -1209,6 +1374,81 @@ describe('FeishuChannel', () => { expect(cache.has('id_25')).toBe(true); expect(cache.has('id_524')).toBe(true); }); + + it('re-hydrates persisted labels after in-lifetime cache eviction', async () => { + const observe = vi.fn(); + const list = vi.fn( + (): ObservedChannelContactGraph => ({ + users: [ + { + channelName: 'test', + id: 'ou_user', + label: 'Alice', + lastObservedAt: '2026-01-01T00:00:00.000Z', + }, + ], + groups: [], + }), + ); + const { channel, bridge } = createObservedContactChannel(observe, list); + const fetchSpy = vi + .spyOn(global, 'fetch') + .mockImplementation(async (input, init) => { + const url = String(input); + if (url.includes('/contact/v3/users/basic_batch')) { + if (String(init?.body).includes('ou_new')) { + return jsonResponse({ + code: 0, + data: { users: [{ name: 'New User' }] }, + }); + } + return jsonResponse('rate limited', 429); + } + throw new Error(`Unexpected request: ${url}`); + }); + const onMessage = getPrivateMethod<(data: unknown) => void>( + channel, + 'onMessage', + ).bind(channel); + const observedUserName = getPrivateMethod< + (userId: string) => Promise + >(channel, 'observedUserName').bind(channel); + + try { + onMessage(feishuDmMessage('dm_1')); + await vi.waitFor(() => { + expect(observe).toHaveBeenCalledTimes(1); + expect(bridge.prompt).toHaveBeenCalledTimes(1); + }); + expect(observe).toHaveBeenNthCalledWith(1, 'test', { + user: { id: 'ou_user', label: 'Alice' }, + }); + + // Churn the label cache past the cap so the hydrated entry evicts + // when the next successful lookup is cached. + const names = (channel as unknown as Record)[ + 'observedUserNames' + ] as Map; + for (let i = 0; i < 500; i++) { + names.set(`ou_filler_${i}`, `User ${i}`); + } + await expect(observedUserName('ou_new')).resolves.toBe('New User'); + expect(names.has('ou_user')).toBe(false); + + onMessage(feishuDmMessage('dm_2')); + await vi.waitFor(() => { + expect(observe).toHaveBeenCalledTimes(2); + expect(bridge.prompt).toHaveBeenCalledTimes(2); + }); + expect(observe).toHaveBeenNthCalledWith(2, 'test', { + user: { id: 'ou_user', label: 'Alice' }, + }); + expect(list).toHaveBeenCalledTimes(2); + expect(fetchSpy).toHaveBeenCalledTimes(1); + } finally { + fetchSpy.mockRestore(); + } + }); }); describe('extractCardText', () => {