From 028747aa417a97cbead7087f4b7c2beec246e87d Mon Sep 17 00:00:00 2001 From: BaboBen <117555359+BenGuanRan@users.noreply.github.com> Date: Sat, 8 Aug 2026 00:41:37 +0800 Subject: [PATCH] feat(feishu): enrich observed contact labels (#8569) * docs: design feishu observed contact enrichment * docs: add Chinese Feishu enrichment design * feat(feishu): enrich observed contact labels * fix(feishu): preserve enriched contact labels * fix(feishu): harden observed-contact label enrichment lifecycle * 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 * 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. --------- Co-authored-by: qwen-code-dev-bot Co-authored-by: Shaojin Wen Co-authored-by: qwen-code-ci-bot --- ...eishu-observed-contact-label-enrichment.md | 114 ++ ...eishu-observed-contact-label-enrichment.md | 290 ++++ docs/users/features/channels/feishu.md | 9 + .../channels/base/src/ChannelBase.test.ts | 62 + packages/channels/base/src/ChannelBase.ts | 29 +- packages/channels/feishu/src/FeishuAdapter.ts | 253 +++- packages/channels/feishu/src/adapter.test.ts | 1192 +++++++++++++++++ .../commands/channel/daemon-worker.test.ts | 9 + .../cli/src/commands/channel/daemon-worker.ts | 9 +- 9 files changed, 1957 insertions(+), 10 deletions(-) create mode 100644 docs/design/feishu-observed-contact-label-enrichment.md 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 new file mode 100644 index 0000000000..fd37444f76 --- /dev/null +++ b/docs/design/feishu-observed-contact-label-enrichment.md @@ -0,0 +1,114 @@ +# Feishu observed-contact label enrichment + +## 中文说明 + +### 目标 + +在不延迟飞书入站消息处理的前提下，用真实用户名和群名补全观测联系人 +的 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 重 +启为止，因 tenant token 获取失败而未能发出的请求以及收到 401 的请求 +保持可重试，401 同时使缓存的 tenant token 失效。daemon 重启后第一条 +入站消息会从已持久化的观测联系人记录回填此前解析出的名称，避免已知 +名称在重新查询期间被回退为原始 ID。查询请求的 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 后触发且不会被等待。 +飞书适配器测试验证成功补全、进程内去重、后续消息复用名称、daemon 重启 +后回填已解析名称，以及失败静默时原始 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 + +`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 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 +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 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 + +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, 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/docs/plans/feishu-observed-contact-label-enrichment.md b/docs/plans/feishu-observed-contact-label-enrichment.md new file mode 100644 index 0000000000..793bf14b91 --- /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/docs/users/features/channels/feishu.md b/docs/users/features/channels/feishu.md index 0ed38160e1..cc7fb305ea 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/base/src/ChannelBase.test.ts b/packages/channels/base/src/ChannelBase.test.ts index 4abafbfaa2..c474057f35 100644 --- a/packages/channels/base/src/ChannelBase.test.ts +++ b/packages/channels/base/src/ChannelBase.test.ts @@ -916,6 +916,68 @@ 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('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/base/src/ChannelBase.ts b/packages/channels/base/src/ChannelBase.ts index b97ea18ed9..d09c15131f 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, @@ -230,6 +231,8 @@ export interface ChannelBaseOptions { channelName: string, observation: ObservedChannelContactObservation, ): void | Promise; + /** Read persisted observations so adapters can hydrate label caches. */ + list?(): ObservedChannelContactGraph; }; } @@ -4927,7 +4930,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) @@ -4968,6 +4971,29 @@ 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); } @@ -5000,6 +5026,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 92b7aaf4ca..f9ab0755c9 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'; @@ -83,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; @@ -142,6 +149,24 @@ 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< + string, + Promise + >(); + private readonly observedChatLookups = new Map< + string, + Promise + >(); + private readonly observedContactWrites = new Map< + string, + { senderName: string; chatName: string | undefined } + >(); + private hydratedObservedNames = false; private collapsible: boolean; private collapsibleThreshold: number; @@ -621,21 +646,27 @@ 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 (!options?.silent) this.tokenRefreshHasCoreWaiters = true; if (this.tokenRefreshPromise) return this.tokenRefreshPromise; this.tokenRefreshPromise = this.refreshToken(); try { return await this.tokenRefreshPromise; } finally { this.tokenRefreshPromise = undefined; + this.tokenRefreshHasCoreWaiters = false; } } 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`, @@ -651,9 +682,11 @@ export class FeishuChannel extends ChannelBase { ); if (!resp.ok) { - process.stderr.write( - `[Feishu:${this.name}] getTenantAccessToken failed: HTTP ${resp.status}\n`, - ); + if (this.tokenRefreshHasCoreWaiters) { + process.stderr.write( + `[Feishu:${this.name}] getTenantAccessToken failed: HTTP ${resp.status}\n`, + ); + } if (resp.status === 401) this.tokenCache = undefined; return undefined; } @@ -669,13 +702,213 @@ export class FeishuChannel extends ChannelBase { }; return this.tokenCache.token; } catch (err) { - process.stderr.write( - `[Feishu:${this.name}] getTenantAccessToken error: ${err}\n`, - ); + if (this.tokenRefreshHasCoreWaiters) { + process.stderr.write( + `[Feishu:${this.name}] getTenantAccessToken error: ${err}\n`, + ); + } return undefined; } } + private hydrateObservedNames(): void { + if (this.hydratedObservedNames) return; + this.hydratedObservedNames = true; + const graph = this.persistedObservedContacts(); + if (!graph) return; + // 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) return; + const current = best.get(id); + if (!current || at >= current.at) best.set(id, { label, at }); + }; + for (const user of graph.users) { + consider(newestUser, user.id, user.label, user.lastObservedAt); + } + for (const group of graph.groups) { + consider(newestChat, group.id, group.label, group.lastObservedAt); + for (const member of group.users) { + 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): 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 { + 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', + headers: { + Authorization: `Bearer ${token}`, + 'Content-Type': 'application/json', + }, + body: JSON.stringify({ user_ids: [userId] }), + signal: AbortSignal.timeout(15_000), + }, + ), + extractName: (body) => { + const data = body as { + code?: number; + data?: { users?: Array<{ name?: string }> }; + }; + return data.code === 0 ? data.data?.users?.[0]?.name : undefined; + }, + }); + } + + private observedChatName(chatId: string): Promise { + 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 cached = options.names.get(options.id); + if (cached) return Promise.resolve(cached); + const existing = options.lookups.get(options.id); + if (existing) return existing; + + const lookup = (async () => { + try { + const token = await this.getTenantAccessToken({ silent: true }); + if (!token) { + options.lookups.delete(options.id); + return undefined; + } + + const response = await options.request(token); + if (!response.ok) { + if (response.status === 401) { + this.tokenCache = undefined; + options.lookups.delete(options.id); + } + return undefined; + } + + 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); + // 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; + } + })(); + options.lookups.set(options.id, lookup); + this.capObservedCache(options.lookups); + return lookup; + } + + protected override onObservedContact(envelope: Envelope): void { + this.observedContactWrites.set(this.observedContactKey(envelope), { + senderName: envelope.senderName, + chatName: envelope.chatName, + }); + this.capObservedCache(this.observedContactWrites); + 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), + envelope.isGroup + ? this.observedChatName(envelope.chatId) + : 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 ( + persistedLabels && + persistedLabels.senderName === nextLabels.senderName && + persistedLabels.chatName === nextLabels.chatName + ) { + return; + } + this.observedContactWrites.set(key, nextLabels); + this.capObservedCache(this.observedContactWrites); + await this.recordObservedContact({ + ...envelope, + ...(senderName ? { senderName } : {}), + ...(chatName ? { chatName } : {}), + }); + } + async sendMessage(chatId: string, text: string): Promise { await this.sendMessageInternal(chatId, text, false); } @@ -1951,6 +2184,9 @@ 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; // Parse message content const content = this.extractContent(msg.message_type, msg.content); @@ -1995,8 +2231,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 97de4df1f0..dc159f6670 100644 --- a/packages/channels/feishu/src/adapter.test.ts +++ b/packages/channels/feishu/src/adapter.test.ts @@ -20,9 +20,11 @@ vi.mock('@larksuiteoapi/node-sdk', async (importOriginal) => { import { FeishuChannel } from './FeishuAdapter.js'; import type { ChannelAgentBridge, + ChannelBaseOptions, ChannelConfig, ChannelProactiveDeliveryError, ChannelTaskLifecycleEvent, + ObservedChannelContactGraph, SessionTarget, } from '@qwen-code/channel-base'; @@ -77,11 +79,105 @@ 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'], + list?: NonNullable['list'], +): { + channel: ObservedContactFeishuChannel; + bridge: ChannelAgentBridge; +} { + const bridge = createMockBridge(); + const channel = new ObservedContactFeishuChannel( + 'test', + createConfig({ + blockStreaming: 'on', + groupPolicy: 'open', + groups: { '*': { requireMention: false } }, + }), + bridge, + { observedContacts: { observe, ...(list ? { list } : {}) } }, + ); + 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', + }, + }; +} + +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; } +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', () => { @@ -259,6 +355,1102 @@ 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(); + } + }); + + 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); + expect(stderrSpy).not.toHaveBeenCalled(); + } 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'); + 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); + expect(stderrSpy).not.toHaveBeenCalled(); + } 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('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); + 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); + // 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); + 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(); + } + }); + + 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('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( + (): 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); + }); + + 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', () => { let channel: FeishuChannel; let extractCardText: (card: Record) => string | undefined; diff --git a/packages/cli/src/commands/channel/daemon-worker.test.ts b/packages/cli/src/commands/channel/daemon-worker.test.ts index fc903cdb2e..3fa19d874b 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 6067633a23..3aa6d3fa95 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 } : {}), }),