diff --git a/docs/design/2026-07-17-observed-channel-delivery-targets.md b/docs/design/2026-07-17-observed-channel-delivery-targets.md new file mode 100644 index 00000000000..1e304520b9c --- /dev/null +++ b/docs/design/2026-07-17-observed-channel-delivery-targets.md @@ -0,0 +1,137 @@ +# Workspace-scoped observed channel contacts + +## Problem + +Daemon-managed channel workers receive platform user, group, and topic identifiers on inbound messages, but the identifiers are transient. Authenticated workspace clients need a read API that lists recently observed IM contacts so a user can select a complete platform delivery target without manually finding or retyping identifiers. + +## Scope + +This change observes accepted inbound messages, persists a bounded relationship graph per daemon workspace, and returns complete platform identifiers for DingTalk, Feishu, Telegram, and WeCom channels. + +It does not change webhook configuration or proactive delivery, query a platform directory, claim to return complete group membership, observe bot output, or backfill historical traffic. Standalone `qwen channel start` is unchanged. + +## Ownership and persistence + +The daemon workspace runtime owns the registry: + +```text +$QWEN_HOME/channels/daemon//observed-contacts.json +``` + +`QWEN_HOME` is process-level, but `` partitions data by canonical workspace path. The registry is not stored in the workspace checkout and is not shared as one process-global graph. Its directory uses mode `0700` where supported; the atomic JSON file uses mode `0600`. + +The registry stores at most 500 relationship observations across all channels and conversations in the workspace. Each observation contains `channelName`, a user identity, an optional group identity, an optional topic identity, and `lastObservedAt`. The deduplication key is `[channelName, user.id, group?.id, topic?.id]`. A noisy conversation can therefore evict older observations from another conversation. Observations older than the maximum 365-day readable window are removed on the next accepted write. + +## Observation boundary + +Recording occurs after the shared inbound preflight accepts a real IM message and before command or Agent handling begins. Direct/group policy, mention, sender allowlist, and pairing rejection therefore happen before persistence. + +The same `Envelope` object is recorded at most once. A later message refreshes the matching relationship timestamp and labels. Persistence is best-effort: a sanitized error is logged without identifiers, and accepted message handling continues. + +The registry never stores message text, message IDs, attachments, payloads, credentials, webhook requests, proactive sends, or bot output. + +## Relationship model + +```ts +interface ObservedChannelContactObservation { + user: { id: string; label: string }; + group?: { id: string; label: string }; + topic?: { id: string; label: string }; +} +``` + +- A direct message records a top-level user from the complete platform `senderId`. +- A group message records the group from the complete platform `chatId` and the observed user inside that group. +- A threaded group message also records the topic from `threadId` and the observed user inside that topic. +- A user seen only in groups does not appear in top-level `users`. If the same user also sends a direct message, it appears both at the top level and under the relevant groups. +- `groups[].users` and `groups[].topics[].users` mean users observed in those conversations. They are not authoritative platform membership lists. +- Sender labels use the sanitized inbound display name, falling back to the complete user ID. The current common envelope has no portable group/topic display name, so those labels fall back to their complete IDs. + +Feishu maps `root_id` to `threadId`; Telegram maps `message_thread_id` to `threadId`. Current DingTalk and WeCom envelopes do not expose a stable topic identifier, so their observations stop at the group level. + +## Freshness + +People, conversations, and relationships change. The read API filters observations rather than presenting the registry as permanent truth: + +- default freshness: seven days; +- caller override: `freshWithinSeconds`, from 1 second through 365 days; +- user, group-user, topic-user, group, and topic timestamps are derived independently from recent observations; +- passive observation cannot immediately detect a leave, deletion, or rename that produces no new message, so stale relationships disappear only when they exceed the requested window. + +## Read API + +Primary workspace: + +```http +GET /workspace/channel/observed-contacts?freshWithinSeconds=604800 +Authorization: Bearer +``` + +Selected registered workspace: + +```http +GET /workspaces/:workspace/channel/observed-contacts?freshWithinSeconds=604800 +Authorization: Bearer +``` + +Example: + +```json +{ + "users": [ + { + "channelName": "feishu-main", + "label": "Example User", + "id": "ou_complete_user_id", + "lastObservedAt": "2026-07-17T08:00:00.000Z" + } + ], + "groups": [ + { + "channelName": "feishu-main", + "label": "oc_complete_chat_id", + "id": "oc_complete_chat_id", + "lastObservedAt": "2026-07-17T08:05:00.000Z", + "users": [ + { + "label": "Example User", + "id": "ou_complete_user_id", + "lastObservedAt": "2026-07-17T08:05:00.000Z" + } + ], + "topics": [ + { + "label": "om_complete_root_id", + "id": "om_complete_root_id", + "lastObservedAt": "2026-07-17T08:05:00.000Z", + "users": [ + { + "label": "Example User", + "id": "ou_complete_user_id", + "lastObservedAt": "2026-07-17T08:05:00.000Z" + } + ] + } + ] + } + ] +} +``` + +Responses use `Cache-Control: no-store`. The primary route reads only the primary workspace partition. The qualified route requires an exact registered, trusted runtime and never falls back to primary for unknown, untrusted, bootstrapping, draining, or removed workspaces. + +A missing registry returns an empty graph. Malformed data returns a sanitized `500` with code `channel_observed_contacts_unavailable`. Delete the workspace's `observed-contacts.json` file to reset a malformed or unsupported registry; accepted traffic recreates it. Invalid freshness returns `400 invalid_freshness`. + +Clients discover the route through the `workspace_channel_observed_contacts` serve capability. The route is read-only and is registered after daemon bearer authentication. + +## Compatibility + +Webhook parsing, requests, target resolution, and delivery are identical to `main`. This API only exposes observed identifiers; callers decide how to use them. The registry begins at schema version 1 because the earlier opaque-reference prototype was never released. + +## Test strategy + +- Base-channel tests cover the preflight boundary, topic normalization, Envelope deduplication, and non-blocking persistence failures. +- Store tests cover direct-versus-group semantics, group/topic relationships, freshness, refreshes, bounds, permissions, and malformed data. +- Route tests cover complete identifiers, no-store responses, freshness validation, exact workspace ownership, and sanitized failures. +- Server tests cover bearer authentication and capability advertisement. +- Webhook regression tests verify no behavior differs from `main`. diff --git a/docs/plans/2026-07-17-observed-channel-delivery-targets.md b/docs/plans/2026-07-17-observed-channel-delivery-targets.md new file mode 100644 index 00000000000..e75ea146b6f --- /dev/null +++ b/docs/plans/2026-07-17-observed-channel-delivery-targets.md @@ -0,0 +1,70 @@ +# Observed Channel Contacts 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. + +**Goal:** Expose a fresh, workspace-scoped graph of dynamically observed direct users, groups, topics, and their observed users with complete platform identifiers. + +**Architecture:** `ChannelBase` normalizes accepted inbound envelopes into relationship observations. The daemon worker writes them to a bounded workspace-partitioned JSON registry. Authenticated read-only routes derive a fresh graph for the exact workspace runtime. Webhook behavior remains unchanged. + +## Constraints + +- Persist under `$QWEN_HOME/channels/daemon//observed-contacts.json`. +- Record after inbound preflight and before command or Agent handling. +- Record each Envelope once; never record rejected input, bot output, proactive sends, or webhook traffic. +- Return complete IDs and sanitized/fallback labels. +- Top-level `users` contains direct-message users only. +- `groups[].users` and `groups[].topics[].users` are observed relationships, not authoritative membership. +- Default freshness is seven days; accept `freshWithinSeconds` from 1 second through 365 days. +- Keep at most 500 most-recent relationship observations. +- Preserve exact workspace ownership and never fall back on qualified routes. +- Do not change webhook configuration, requests, or delivery. + +## Task 1: Base observation contract + +**Files:** `packages/channels/base/src/types.ts`, `ChannelBase.ts`, `ChannelBase.test.ts`, `index.ts` + +- [x] Add identity, observation, graph, group, topic, and related-user types. +- [x] Add `observedContacts.observe` to `ChannelBaseOptions`. +- [x] Normalize `senderId`, `senderName`, `chatId`, and `threadId` after successful preflight. +- [x] Deduplicate the same Envelope object and keep persistence failures non-blocking. +- [x] Cover direct, group, topic, rejection, pairing, duplicate Envelope, and failure cases. + +## Task 2: Workspace relationship store + +**Files:** `packages/cli/src/commands/channel/observed-contact-store.ts`, `observed-contact-store.test.ts`, `daemon-worker.ts`, `daemon-worker.test.ts` + +- [x] Persist version 1 relationship observations with atomic mode-`0600` writes. +- [x] Deduplicate by channel, user, group, and topic; refresh labels and timestamps. +- [x] Derive direct users separately from group and topic relationships. +- [x] Filter stale observations and expose independent relationship timestamps. +- [x] Enforce validation and the 500-observation bound. +- [x] Wire daemon-managed channels to the workspace-partitioned store. + +## Task 3: Authenticated dynamic-observation API + +**Files:** `packages/cli/src/serve/routes/workspace-channel-observed-contacts.ts`, its test, `server.ts`, `server.test.ts`, `capabilities.ts` + +- [x] Add singular and qualified `/channel/observed-contacts` GET routes. +- [x] Parse `freshWithinSeconds`, default to seven days, and reject invalid values. +- [x] Return `{users, groups}` with nested group/topic users and complete IDs. +- [x] Add `Cache-Control: no-store` and sanitized failure responses. +- [x] Require exact trusted workspace resolution on qualified routes. +- [x] Advertise `workspace_channel_observed_contacts`. +- [x] Run the focused server authentication and capability tests. + +## Task 4: Remove prototype webhook integration + +**Files:** `ChannelWebhookTask.ts`, `ChannelBase.ts`, channel config parsing/tests, channel overview documentation + +- [x] Remove observed-reference webhook config and resolution. +- [x] Restore concrete webhook behavior to `origin/main`. +- [x] Verify webhook production files have no diff from `origin/main`. +- [x] Replace prototype documentation with the observed-contacts API. + +## Task 5: Verify and publish + +- [x] Run Prettier on changed files. +- [x] Run focused base, CLI store, daemon-worker, route, server, runtime, and webhook tests. +- [x] Run `npm run build && npm run typecheck`. +- [x] Audit the full diff twice, including untracked files. +- [x] Commit, push `feat/channel-observed-targets`, and update Draft PR #7109. diff --git a/docs/users/features/channels/overview.md b/docs/users/features/channels/overview.md index 10d363ad545..544d9a201e2 100644 --- a/docs/users/features/channels/overview.md +++ b/docs/users/features/channels/overview.md @@ -481,6 +481,48 @@ Example channel config: For DingTalk, set `isGroup` explicitly on every target. A direct-message target uses the DingTalk user ID as `chatId` with `isGroup: false`; a group target uses the group `openConversationId` with `isGroup: true`. Other adapters may require their own proactive target shape. +Daemon-managed DingTalk, Feishu, Telegram, and WeCom channels dynamically observe contacts from authorized inbound messages. List contacts observed in the primary workspace during the default seven-day freshness window: + +```bash +curl -H "Authorization: Bearer $QWEN_SERVER_TOKEN" \ + http://127.0.0.1:4170/workspace/channel/observed-contacts +``` + +Use `GET /workspaces/:workspace/channel/observed-contacts` to select another registered, trusted workspace. Add `?freshWithinSeconds=N` to choose a window from one second through 365 days. The daemon advertises this API with the `workspace_channel_observed_contacts` capability. + +The response returns complete platform IDs and labels. Each `lastObservedAt` is a canonical ISO 8601 UTC timestamp with millisecond precision; clients can convert it to the user's local time zone for display. Top-level `users` contains users observed in direct messages. `groups` contains observed group conversations, `groups[].users` contains users observed in each group, and `groups[].topics[].users` contains users observed in Feishu or Telegram topics: + +```json +{ + "users": [ + { + "channelName": "feishu-main", + "label": "Example User", + "id": "ou_complete_user_id", + "lastObservedAt": "2026-07-17T08:00:00.000Z" + } + ], + "groups": [ + { + "channelName": "feishu-main", + "label": "oc_complete_chat_id", + "id": "oc_complete_chat_id", + "lastObservedAt": "2026-07-17T08:05:00.000Z", + "users": [ + { + "label": "Example User", + "id": "ou_complete_user_id", + "lastObservedAt": "2026-07-17T08:05:00.000Z" + } + ], + "topics": [] + } + ] +} +``` + +These nested users are observed participants, not authoritative group membership. Only messages that pass direct/group, mention, sender, and pairing gates are recorded. Repeated observations refresh labels and timestamps; passive observation cannot detect a leave or deletion until the relationship becomes stale. Message content is never stored. The bounded registry lives under `$QWEN_HOME/channels/daemon//observed-contacts.json`, outside the workspace checkout and partitioned per workspace. Its 500-observation limit is shared by all channels and conversations in that workspace, and observations older than 365 days are removed on the next accepted write. If the registry becomes malformed or uses an unsupported version, delete that file to reset it; accepted traffic recreates it. Webhook configuration and delivery are unchanged. + Start `qwen serve` with the channel worker enabled: ```bash diff --git a/integration-tests/cli/qwen-serve-routes.test.ts b/integration-tests/cli/qwen-serve-routes.test.ts index 8e11e9a548e..472d31cabb8 100644 --- a/integration-tests/cli/qwen-serve-routes.test.ts +++ b/integration-tests/cli/qwen-serve-routes.test.ts @@ -379,6 +379,7 @@ describe('qwen serve — capabilities envelope', () => { 'session_branch', 'workspace_reload', 'channel_control', + 'workspace_channel_observed_contacts', 'persistent_workspace_registration', 'workspace_runtime_removal', 'workspace_qualified_rest_core', diff --git a/packages/channels/base/src/ChannelBase.test.ts b/packages/channels/base/src/ChannelBase.test.ts index 4385f200557..54a794e358f 100644 --- a/packages/channels/base/src/ChannelBase.test.ts +++ b/packages/channels/base/src/ChannelBase.test.ts @@ -116,6 +116,12 @@ class TestChannel extends ChannelBase { this.proactiveTargets.push(target); } + async processAfterAdapterPreflight(envelope: Envelope): Promise { + if (await this.preflightInbound(envelope)) { + await this.processInbound(envelope); + } + } + enableCancelCommand(): void { this.registerCancelCommand(); } @@ -650,6 +656,123 @@ describe('ChannelBase', () => { await ch.handleInbound(envelope()); expect(bridge.prompt).toHaveBeenCalled(); }); + + it('observes a user after inbound gates pass', async () => { + const observe = vi.fn(); + const ch = createChannel({}, { observedContacts: { observe } }); + + await ch.handleInbound(envelope()); + + expect(observe).toHaveBeenCalledWith('test-chan', { + user: { id: 'user1', label: 'User 1' }, + }); + 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 } }); + + await ch.handleInbound(envelope({ senderName: '\u0000\n' })); + + expect(observe).toHaveBeenCalledWith('test-chan', { + user: { id: 'user1', label: 'user1' }, + }); + }); + + it('observes a group, topic, and user relationship after adapter preflight', async () => { + const observe = vi.fn(); + const ch = createChannel( + { groupPolicy: 'open' }, + { observedContacts: { observe } }, + ); + + await ch.processAfterAdapterPreflight( + envelope({ + chatId: 'group-1', + threadId: 'topic-1', + isGroup: true, + isMentioned: true, + }), + ); + + expect(observe).toHaveBeenCalledWith('test-chan', { + user: { id: 'user1', label: 'User 1' }, + group: { id: 'group-1', label: 'group-1' }, + topic: { id: 'topic-1', label: 'topic-1' }, + }); + }); + + it('records the same inbound envelope only once', async () => { + const observe = vi.fn(); + const ch = createChannel({}, { observedContacts: { observe } }); + const message = envelope(); + + await ch.processAfterAdapterPreflight(message); + await ch.processAfterAdapterPreflight(message); + + expect(observe).toHaveBeenCalledTimes(1); + }); + + it.each([ + { + name: 'group policy', + config: {}, + message: { isGroup: true }, + }, + { + name: 'DM policy', + config: { dmPolicy: 'disabled' as const }, + message: {}, + }, + { + name: 'sender policy', + config: { + senderPolicy: 'allowlist' as const, + allowedUsers: ['other'], + }, + message: {}, + }, + ])( + 'does not observe contacts rejected by $name', + async ({ config, message }) => { + const observe = vi.fn(); + const ch = createChannel(config, { observedContacts: { observe } }); + + await ch.handleInbound(envelope(message)); + + expect(observe).not.toHaveBeenCalled(); + }, + ); + + it('does not observe senders waiting for pairing approval', async () => { + const observe = vi.fn(); + const ch = createChannel( + { senderPolicy: 'pairing', allowedUsers: [] }, + { observedContacts: { observe } }, + ); + + await ch.handleInbound(envelope({ senderId: 'stranger' })); + + expect(observe).not.toHaveBeenCalled(); + expect(bridge.prompt).not.toHaveBeenCalled(); + }); + + it('continues inbound processing when contact observation fails', async () => { + const observe = vi.fn().mockRejectedValue(new Error('private-id leaked')); + const stderr = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + const ch = createChannel({}, { observedContacts: { observe } }); + + await expect(ch.handleInbound(envelope())).resolves.toBeUndefined(); + + expect(bridge.prompt).toHaveBeenCalled(); + const logged = stderr.mock.calls.map((call) => String(call[0])).join(''); + stderr.mockRestore(); + expect(logged).toContain('observed contact persistence failed'); + expect(logged).not.toContain('private-id'); + }); }); describe('permission relay', () => { diff --git a/packages/channels/base/src/ChannelBase.ts b/packages/channels/base/src/ChannelBase.ts index 09c8ebcbc91..7135f12856e 100644 --- a/packages/channels/base/src/ChannelBase.ts +++ b/packages/channels/base/src/ChannelBase.ts @@ -12,6 +12,7 @@ import type { ChannelTaskLifecycleEvent, DispatchMode, Envelope, + ObservedChannelContactObservation, SanitizedToolCallEvent, SessionTarget, } from './types.js'; @@ -150,6 +151,12 @@ export interface ChannelBaseOptions { registerBridgeEvents?: boolean; groupHistoryPath?: string; loopController?: ChannelLoopController; + observedContacts?: { + observe( + channelName: string, + observation: ObservedChannelContactObservation, + ): void | Promise; + }; } export interface ChannelLoopController { @@ -279,6 +286,8 @@ export abstract class ChannelBase { private readonly memoryIntentClassifier?: ChannelMemoryIntentClassifier; private groupHistory: GroupHistoryStore; private readonly loopController?: ChannelLoopController; + private readonly observedContacts?: ChannelBaseOptions['observedContacts']; + private readonly observedContactEnvelopes = new WeakSet(); private instructedSessions: Set = new Set(); private commands: Map = new Map(); /** Per-session promise chain to serialize prompt + send (followup mode). */ @@ -472,6 +481,7 @@ export abstract class ChannelBase { ), ); this.loopController = options?.loopController; + this.observedContacts = options?.observedContacts; this.groupGate = new GroupGate(config.groupPolicy, config.groups); this.dmGate = new DmGate(config.dmPolicy); @@ -3796,6 +3806,40 @@ export abstract class ChannelBase { await this.processInbound(envelope); } + private async recordObservedContact(envelope: Envelope): Promise { + if (!this.observedContacts) return; + const sanitizedSenderName = envelope.senderName + ? sanitizeSenderName(envelope.senderName) + : ''; + const userLabel = + sanitizedSenderName === 'unknown' + ? envelope.senderId + : sanitizedSenderName || envelope.senderId; + const observation: ObservedChannelContactObservation = { + user: { id: envelope.senderId, label: userLabel }, + ...(envelope.isGroup + ? { + group: { id: envelope.chatId, label: envelope.chatId }, + ...(envelope.threadId + ? { + topic: { + id: envelope.threadId, + label: envelope.threadId, + }, + } + : {}), + } + : {}), + }; + try { + await this.observedContacts.observe(this.name, observation); + } catch { + process.stderr.write( + `[Channel:${sanitizeLogText(this.name, 80)}] observed contact persistence failed.\n`, + ); + } + } + protected markPreflighted(envelope: Envelope): void { this.preflightedEnvelopes.add(envelope); } @@ -3813,6 +3857,10 @@ export abstract class ChannelBase { 'processInbound called without a successful preflightInbound check.', ); } + if (this.observedContacts && !this.observedContactEnvelopes.has(envelope)) { + this.observedContactEnvelopes.add(envelope); + await this.recordObservedContact(envelope); + } let memoryIntent: ResolvedChannelMemoryIntent | null = parseChannelMemoryIntent(envelope.text); diff --git a/packages/channels/base/src/index.ts b/packages/channels/base/src/index.ts index a2908576986..a91a2ac3f67 100644 --- a/packages/channels/base/src/index.ts +++ b/packages/channels/base/src/index.ts @@ -92,6 +92,13 @@ export type { Envelope, GroupConfig, GroupPolicy, + ObservedChannelIdentity, + ObservedChannelContactObservation, + ObservedChannelContact, + ObservedChannelRelatedContact, + ObservedChannelTopic, + ObservedChannelGroup, + ObservedChannelContactGraph, SanitizedToolCallEvent, SenderPolicy, SessionScope, diff --git a/packages/channels/base/src/types.ts b/packages/channels/base/src/types.ts index fd465ed8c9e..c204cc5fc7c 100644 --- a/packages/channels/base/src/types.ts +++ b/packages/channels/base/src/types.ts @@ -135,6 +135,40 @@ export interface SessionTarget { isGroup?: boolean; } +export interface ObservedChannelIdentity { + id: string; + label: string; +} + +export interface ObservedChannelContactObservation { + user: ObservedChannelIdentity; + group?: ObservedChannelIdentity; + topic?: ObservedChannelIdentity; +} + +export interface ObservedChannelContact extends ObservedChannelIdentity { + channelName: string; + lastObservedAt: string; +} + +export interface ObservedChannelRelatedContact extends ObservedChannelIdentity { + lastObservedAt: string; +} + +export interface ObservedChannelTopic extends ObservedChannelRelatedContact { + users: ObservedChannelRelatedContact[]; +} + +export interface ObservedChannelGroup extends ObservedChannelContact { + users: ObservedChannelRelatedContact[]; + topics: ObservedChannelTopic[]; +} + +export interface ObservedChannelContactGraph { + users: ObservedChannelContact[]; + groups: ObservedChannelGroup[]; +} + export interface ChannelTaskLifecycleBase { channelName: string; chatId: string; diff --git a/packages/cli/src/commands/channel/daemon-worker.test.ts b/packages/cli/src/commands/channel/daemon-worker.test.ts index fb14be1e506..c91b11c3f55 100644 --- a/packages/cli/src/commands/channel/daemon-worker.test.ts +++ b/packages/cli/src/commands/channel/daemon-worker.test.ts @@ -18,6 +18,17 @@ const mockSessionsPath = vi.hoisted(() => vi.fn(() => '/tmp/sessions.json')); const mockDaemonSessionRoutesPath = vi.hoisted(() => vi.fn(() => '/tmp/qwen/channels/daemon/workspace-hash/routes.json'), ); +const mockDaemonObservedContactsPath = vi.hoisted(() => + vi.fn( + () => '/tmp/qwen/channels/daemon/workspace-hash/observed-contacts.json', + ), +); +const mockObserveContact = vi.hoisted(() => vi.fn()); +const mockObservedContactStore = vi.hoisted(() => + vi.fn(() => ({ + observe: mockObserveContact, + })), +); const mockLoadSettings = vi.hoisted(() => vi.fn((_cwd?: string, _opts?: unknown) => ({ merged: { proxy: 'http://settings-proxy:8080' as string | undefined }, @@ -156,6 +167,7 @@ vi.mock('./proxy.js', () => ({ vi.mock('./runtime.js', () => ({ createChannel: mockCreateChannel, + daemonObservedContactsPath: mockDaemonObservedContactsPath, daemonSessionRoutesPath: mockDaemonSessionRoutesPath, loadChannelsConfig: mockLoadChannelsConfig, loadChannelsFromExtensions: mockLoadChannelsFromExtensions, @@ -167,6 +179,10 @@ vi.mock('./runtime.js', () => ({ sessionsPath: mockSessionsPath, })); +vi.mock('./observed-contact-store.js', () => ({ + ObservedChannelContactStore: mockObservedContactStore, +})); + vi.mock('@qwen-code/channel-base', () => ({ DaemonChannelBridge: mockDaemonChannelBridge, sanitizeLogText: mockSanitizeLogText, @@ -689,8 +705,26 @@ describe('runChannelDaemonWorker', () => { memoryIntentClassifier: expect.objectContaining({ classifyChannelMemoryIntent: expect.any(Function), }), + observedContacts: { + observe: expect.any(Function), + }, }), ); + expect(mockDaemonObservedContactsPath).toHaveBeenCalledWith('/workspace'); + expect(mockObservedContactStore).toHaveBeenCalledWith( + '/tmp/qwen/channels/daemon/workspace-hash/observed-contacts.json', + ); + const channelOptions = mockCreateChannel.mock.calls[0]![3] as { + observedContacts: { + observe(channelName: string, observation: unknown): unknown; + }; + }; + const observation = { + user: { id: '42', label: 'Ada' }, + group: { id: 'group-1', label: 'group-1' }, + }; + channelOptions.observedContacts.observe('telegram', observation); + expect(mockObserveContact).toHaveBeenCalledWith('telegram', observation); 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 e487ad99cc2..0b092b373c2 100644 --- a/packages/cli/src/commands/channel/daemon-worker.ts +++ b/packages/cli/src/commands/channel/daemon-worker.ts @@ -51,6 +51,7 @@ import { writeStderrLine, writeStdoutLine } from '../../utils/stdioHelpers.js'; import { resolveProxyUrl } from './proxy.js'; import { createChannel, + daemonObservedContactsPath, daemonSessionRoutesPath, loadChannelsConfig, loadChannelsFromExtensions, @@ -62,6 +63,7 @@ import { type ParsedChannel, } from './runtime.js'; import { BridgeChannelMemoryIntentClassifier } from './memory-intent-classifier.js'; +import { ObservedChannelContactStore } from './observed-contact-store.js'; const SESSION_SHELL_COMMAND_FEATURE = 'session_shell_command'; const MAX_ACTIVE_WEBHOOK_TASKS = 16; @@ -420,6 +422,9 @@ export async function runChannelDaemonWorker( ); validateChannelWorkspaces(parsed, daemonWorkspace); const modelServiceId = selectFirstModel(parsed, 'Daemon worker'); + const observedContacts = new ObservedChannelContactStore( + daemonObservedContactsPath(daemonWorkspace), + ); const bridge = new DaemonChannelBridge({ cwd: daemonWorkspace, @@ -498,6 +503,11 @@ export async function runChannelDaemonWorker( bridgeFacade, config.cwd, ), + observedContacts: { + observe: (channelName, observation) => { + observedContacts.observe(channelName, observation); + }, + }, }), startupSignal, ), diff --git a/packages/cli/src/commands/channel/observed-contact-store.test.ts b/packages/cli/src/commands/channel/observed-contact-store.test.ts new file mode 100644 index 00000000000..48d541a6e17 --- /dev/null +++ b/packages/cli/src/commands/channel/observed-contact-store.test.ts @@ -0,0 +1,225 @@ +import { mkdtempSync, readFileSync, statSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { beforeEach, describe, expect, it } from 'vitest'; +import { ObservedChannelContactStore } from './observed-contact-store.js'; + +describe('ObservedChannelContactStore', () => { + let filePath: string; + let now: Date; + + beforeEach(() => { + filePath = join( + mkdtempSync(join(tmpdir(), 'qwen-observed-contacts-')), + 'observed-contacts.json', + ); + now = new Date('2026-07-17T12:00:00.000Z'); + }); + + function createStore(maxObservations = 500): ObservedChannelContactStore { + return new ObservedChannelContactStore(filePath, { + now: () => now, + maxObservations, + }); + } + + it('returns an empty graph when the registry is missing', () => { + expect( + createStore().list({ freshWithinSeconds: 7 * 24 * 60 * 60 }), + ).toEqual({ users: [], groups: [] }); + }); + + it('separates direct users from observed group and topic membership', () => { + const store = createStore(); + store.observe('dingtalk-main', { + user: { id: 'direct-user', label: 'Direct User' }, + }); + store.observe('dingtalk-main', { + user: { id: 'user-1', label: 'User One' }, + group: { id: 'group-1', label: 'group-1' }, + topic: { id: 'topic-1', label: 'topic-1' }, + }); + + expect(store.list({ freshWithinSeconds: 604800 })).toEqual({ + users: [ + { + channelName: 'dingtalk-main', + id: 'direct-user', + label: 'Direct User', + lastObservedAt: '2026-07-17T12:00:00.000Z', + }, + ], + groups: [ + { + channelName: 'dingtalk-main', + id: 'group-1', + label: 'group-1', + lastObservedAt: '2026-07-17T12:00:00.000Z', + users: [ + { + id: 'user-1', + label: 'User One', + lastObservedAt: '2026-07-17T12:00:00.000Z', + }, + ], + topics: [ + { + id: 'topic-1', + label: 'topic-1', + lastObservedAt: '2026-07-17T12:00:00.000Z', + users: [ + { + id: 'user-1', + label: 'User One', + lastObservedAt: '2026-07-17T12:00:00.000Z', + }, + ], + }, + ], + }, + ], + }); + }); + + it('keeps the same user in both direct and group observations', () => { + const store = createStore(); + store.observe('feishu', { + user: { id: 'user-1', label: 'User One' }, + }); + store.observe('feishu', { + user: { id: 'user-1', label: 'User One' }, + group: { id: 'group-1', label: 'group-1' }, + }); + + const graph = store.list({ freshWithinSeconds: 604800 }); + expect(graph.users.map((user) => user.id)).toEqual(['user-1']); + expect(graph.groups[0]?.users.map((user) => user.id)).toEqual(['user-1']); + }); + + it('updates labels and timestamps when a relationship is observed again', () => { + const store = createStore(); + store.observe('telegram', { + user: { id: '42', label: 'Old Name' }, + group: { id: '-100', label: '-100' }, + }); + now = new Date('2026-07-18T12:00:00.000Z'); + store.observe('telegram', { + user: { id: '42', label: 'New Name' }, + group: { id: '-100', label: 'New Group Name' }, + }); + + const graph = store.list({ freshWithinSeconds: 604800 }); + expect(graph.users).toEqual([]); + expect(graph.groups[0]).toMatchObject({ + id: '-100', + label: 'New Group Name', + lastObservedAt: '2026-07-18T12:00:00.000Z', + }); + expect(graph.groups[0]?.users[0]).toMatchObject({ + id: '42', + label: 'New Name', + lastObservedAt: '2026-07-18T12:00:00.000Z', + }); + }); + + it('filters stale users, groups, and group-user relationships', () => { + const store = createStore(); + store.observe('wecom', { + user: { id: 'stale-user', label: 'Stale User' }, + group: { id: 'group-1', label: 'group-1' }, + }); + now = new Date('2026-07-25T12:00:01.000Z'); + + expect(store.list({ freshWithinSeconds: 7 * 24 * 60 * 60 })).toEqual({ + users: [], + groups: [], + }); + }); + + it('keeps the newest bounded relationship observations', () => { + const store = createStore(2); + for (const id of ['1', '2', '3']) { + store.observe('feishu', { user: { id, label: `User ${id}` } }); + now = new Date(now.getTime() + 1000); + } + + expect( + store.list({ freshWithinSeconds: 604800 }).users.map((user) => user.id), + ).toEqual(['3', '2']); + }); + + it('drops observations older than the maximum readable window on write', () => { + const store = createStore(); + store.observe('wecom', { + user: { id: 'stale-user', label: 'Stale User' }, + }); + now = new Date(now.getTime() + 365 * 24 * 60 * 60 * 1000 + 1000); + store.observe('wecom', { + user: { id: 'fresh-user', label: 'Fresh User' }, + }); + + const persisted = JSON.parse(readFileSync(filePath, 'utf8')) as { + observations: Array<{ user: { id: string } }>; + }; + expect(persisted.observations.map((item) => item.user.id)).toEqual([ + 'fresh-user', + ]); + }); + + it('preserves complete IDs while truncating fallback labels', () => { + const store = createStore(); + const longId = 'x'.repeat(300); + + store.observe('telegram', { + user: { id: longId, label: longId }, + group: { id: longId, label: longId }, + }); + + const group = store.list({ freshWithinSeconds: 604800 }).groups[0]; + expect(group?.id).toBe(longId); + expect(group?.label).toBe('x'.repeat(256)); + expect(group?.users[0]?.id).toBe(longId); + expect(group?.users[0]?.label).toBe('x'.repeat(256)); + }); + + it('does not split a Unicode surrogate pair when truncating labels', () => { + const store = createStore(); + + store.observe('feishu', { + user: { + id: 'user-1', + label: `${'x'.repeat(255)}😀suffix`, + }, + }); + + expect(store.list({ freshWithinSeconds: 604800 }).users[0]?.label).toBe( + 'x'.repeat(255), + ); + }); + + it('uses private file permissions where supported', () => { + createStore().observe('telegram', { + user: { id: '1', label: 'User One' }, + }); + + if (process.platform !== 'win32') { + expect(statSync(filePath).mode & 0o777).toBe(0o600); + expect(statSync(join(filePath, '..')).mode & 0o777).toBe(0o700); + } + expect(JSON.parse(readFileSync(filePath, 'utf8'))).toMatchObject({ + version: 1, + }); + }); + + it('rejects malformed or unsupported registries', () => { + writeFileSync(filePath, JSON.stringify({ version: 2, observations: [] })); + expect(() => createStore().list({ freshWithinSeconds: 604800 })).toThrow( + 'Unsupported observed contact registry version', + ); + + writeFileSync(filePath, '{'); + expect(() => createStore().list({ freshWithinSeconds: 604800 })).toThrow( + 'Invalid observed contact registry', + ); + }); +}); diff --git a/packages/cli/src/commands/channel/observed-contact-store.ts b/packages/cli/src/commands/channel/observed-contact-store.ts new file mode 100644 index 00000000000..42f4e27fce6 --- /dev/null +++ b/packages/cli/src/commands/channel/observed-contact-store.ts @@ -0,0 +1,362 @@ +import { chmodSync, existsSync, mkdirSync, readFileSync } from 'node:fs'; +import { dirname } from 'node:path'; +import type { + ObservedChannelContactGraph, + ObservedChannelContactObservation, + ObservedChannelGroup, + ObservedChannelIdentity, + ObservedChannelRelatedContact, + ObservedChannelTopic, +} from '@qwen-code/channel-base'; +import { atomicWriteFileSync } from '@qwen-code/qwen-code-core'; + +const REGISTRY_VERSION = 1; +const MAX_OBSERVATIONS = 500; +const MAX_CHANNEL_NAME_LENGTH = 256; +const MAX_LABEL_LENGTH = 256; +const MAX_ID_LENGTH = 4096; +export const OBSERVED_CONTACT_MAX_FRESH_WITHIN_SECONDS = 365 * 24 * 60 * 60; + +interface PersistedObservedContact { + channelName: string; + user: ObservedChannelIdentity; + group?: ObservedChannelIdentity; + topic?: ObservedChannelIdentity; + lastObservedAt: string; +} + +interface ObservedContactRegistryFile { + version: typeof REGISTRY_VERSION; + observations: PersistedObservedContact[]; +} + +interface ObservedChannelContactStoreOptions { + now?: () => Date; + maxObservations?: number; +} + +interface ListObservedContactsOptions { + freshWithinSeconds: number; +} + +interface MutableObservedGroup extends ObservedChannelGroup { + userMap: Map; + topicMap: Map; +} + +interface MutableObservedTopic extends ObservedChannelTopic { + userMap: Map; +} + +export class ObservedChannelContactStore { + private readonly now: () => Date; + private readonly maxObservations: number; + + constructor( + private readonly filePath: string, + options: ObservedChannelContactStoreOptions = {}, + ) { + this.now = options.now ?? (() => new Date()); + this.maxObservations = options.maxObservations ?? MAX_OBSERVATIONS; + } + + observe( + channelName: string, + observation: ObservedChannelContactObservation, + ): void { + this.validateObservation(channelName, observation); + const observedAt = this.now(); + const next: PersistedObservedContact = { + channelName, + user: this.normalizeIdentity(observation.user), + ...(observation.group + ? { group: this.normalizeIdentity(observation.group) } + : {}), + ...(observation.topic + ? { topic: this.normalizeIdentity(observation.topic) } + : {}), + lastObservedAt: observedAt.toISOString(), + }; + const key = this.observationKey(next); + const retentionCutoff = + observedAt.getTime() - OBSERVED_CONTACT_MAX_FRESH_WITHIN_SECONDS * 1000; + const observations = this.readObservations() + .filter( + (candidate) => + Date.parse(candidate.lastObservedAt) >= retentionCutoff && + this.observationKey(candidate) !== key, + ) + .concat(next) + .sort((a, b) => b.lastObservedAt.localeCompare(a.lastObservedAt)) + .slice(0, this.maxObservations); + this.persist(observations); + } + + list(options: ListObservedContactsOptions): ObservedChannelContactGraph { + const { freshWithinSeconds } = options; + if ( + !Number.isInteger(freshWithinSeconds) || + freshWithinSeconds < 1 || + freshWithinSeconds > OBSERVED_CONTACT_MAX_FRESH_WITHIN_SECONDS + ) { + throw new Error('Invalid observed contact freshness.'); + } + const cutoff = this.now().getTime() - freshWithinSeconds * 1000; + const observations = this.readObservations() + .filter((observation) => Date.parse(observation.lastObservedAt) >= cutoff) + .sort((a, b) => b.lastObservedAt.localeCompare(a.lastObservedAt)); + const users = new Map< + string, + ObservedChannelContactGraph['users'][number] + >(); + const groups = new Map(); + + for (const observation of observations) { + if (!observation.group) { + const key = this.identityKey( + observation.channelName, + observation.user.id, + ); + if (!users.has(key)) { + users.set(key, { + channelName: observation.channelName, + ...observation.user, + lastObservedAt: observation.lastObservedAt, + }); + } + continue; + } + + const groupKey = this.identityKey( + observation.channelName, + observation.group.id, + ); + let group = groups.get(groupKey); + if (!group) { + group = { + channelName: observation.channelName, + ...observation.group, + lastObservedAt: observation.lastObservedAt, + users: [], + topics: [], + userMap: new Map(), + topicMap: new Map(), + }; + groups.set(groupKey, group); + } + + if (!group.userMap.has(observation.user.id)) { + group.userMap.set(observation.user.id, { + ...observation.user, + lastObservedAt: observation.lastObservedAt, + }); + } + + if (observation.topic) { + let topic = group.topicMap.get(observation.topic.id); + if (!topic) { + topic = { + ...observation.topic, + lastObservedAt: observation.lastObservedAt, + users: [], + userMap: new Map(), + }; + group.topicMap.set(observation.topic.id, topic); + } + if (!topic.userMap.has(observation.user.id)) { + topic.userMap.set(observation.user.id, { + ...observation.user, + lastObservedAt: observation.lastObservedAt, + }); + } + } + } + + return { + users: [...users.values()], + groups: [...groups.values()].map((group) => ({ + channelName: group.channelName, + id: group.id, + label: group.label, + lastObservedAt: group.lastObservedAt, + users: [...group.userMap.values()], + topics: [...group.topicMap.values()].map((topic) => ({ + id: topic.id, + label: topic.label, + lastObservedAt: topic.lastObservedAt, + users: [...topic.userMap.values()], + })), + })), + }; + } + + private readObservations(): PersistedObservedContact[] { + if (!existsSync(this.filePath)) return []; + + let parsed: unknown; + try { + parsed = JSON.parse(readFileSync(this.filePath, 'utf-8')); + } catch { + throw new Error('Invalid observed contact registry.'); + } + if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) { + throw new Error('Invalid observed contact registry.'); + } + const record = parsed as Record; + if (record['version'] !== REGISTRY_VERSION) { + throw new Error('Unsupported observed contact registry version.'); + } + if ( + !Array.isArray(record['observations']) || + record['observations'].length > this.maxObservations + ) { + throw new Error('Invalid observed contact registry.'); + } + + const observations = record['observations'].map((raw) => + this.parseObservation(raw), + ); + const keys = new Set(observations.map((item) => this.observationKey(item))); + if (keys.size !== observations.length) { + throw new Error('Invalid observed contact registry.'); + } + return observations; + } + + private parseObservation(raw: unknown): PersistedObservedContact { + if (!raw || typeof raw !== 'object' || Array.isArray(raw)) { + throw new Error('Invalid observed contact registry.'); + } + const record = raw as Record; + const channelName = record['channelName']; + const lastObservedAt = record['lastObservedAt']; + const user = this.parseIdentity(record['user']); + const group = + record['group'] === undefined + ? undefined + : this.parseIdentity(record['group']); + const topic = + record['topic'] === undefined + ? undefined + : this.parseIdentity(record['topic']); + if ( + !this.isBoundedString(channelName, MAX_CHANNEL_NAME_LENGTH) || + typeof lastObservedAt !== 'string' || + !this.isCanonicalTimestamp(lastObservedAt) || + (topic !== undefined && group === undefined) + ) { + throw new Error('Invalid observed contact registry.'); + } + return { + channelName, + user, + ...(group ? { group } : {}), + ...(topic ? { topic } : {}), + lastObservedAt, + }; + } + + private parseIdentity(raw: unknown): ObservedChannelIdentity { + if (!raw || typeof raw !== 'object' || Array.isArray(raw)) { + throw new Error('Invalid observed contact registry.'); + } + const record = raw as Record; + const id = record['id']; + const label = record['label']; + if ( + !this.isBoundedString(id, MAX_ID_LENGTH) || + !this.isBoundedString(label, MAX_LABEL_LENGTH) + ) { + throw new Error('Invalid observed contact registry.'); + } + return { id, label }; + } + + private validateObservation( + channelName: string, + observation: ObservedChannelContactObservation, + ): void { + if ( + !this.isBoundedString(channelName, MAX_CHANNEL_NAME_LENGTH) || + !this.isIdentity(observation.user) || + (observation.group !== undefined && + !this.isIdentity(observation.group)) || + (observation.topic !== undefined && + !this.isIdentity(observation.topic)) || + (observation.topic !== undefined && observation.group === undefined) + ) { + throw new Error('Invalid observed contact observation.'); + } + } + + private isIdentity(value: ObservedChannelIdentity): boolean { + return ( + this.isBoundedString(value.id, MAX_ID_LENGTH) && + this.isBoundedString(value.label, MAX_ID_LENGTH) + ); + } + + private normalizeIdentity( + value: ObservedChannelIdentity, + ): ObservedChannelIdentity { + return { + id: value.id, + label: this.truncateLabel(value.label), + }; + } + + private truncateLabel(value: string): string { + let result = ''; + for (const character of value) { + if (result.length + character.length > MAX_LABEL_LENGTH) break; + result += character; + } + return result; + } + + private observationKey(observation: PersistedObservedContact): string { + return JSON.stringify([ + observation.channelName, + observation.user.id, + observation.group?.id ?? '', + observation.topic?.id ?? '', + ]); + } + + private identityKey(channelName: string, id: string): string { + return JSON.stringify([channelName, id]); + } + + private persist(observations: PersistedObservedContact[]): void { + const dir = dirname(this.filePath); + mkdirSync(dir, { recursive: true, mode: 0o700 }); + try { + chmodSync(dir, 0o700); + } catch { + // Windows and some filesystems do not implement POSIX modes. + } + const data: ObservedContactRegistryFile = { + version: REGISTRY_VERSION, + observations, + }; + atomicWriteFileSync(this.filePath, JSON.stringify(data, null, 2), { + encoding: 'utf-8', + mode: 0o600, + forceMode: true, + noFollow: true, + }); + } + + private isBoundedString(value: unknown, maxLength: number): value is string { + return ( + typeof value === 'string' && value.length > 0 && value.length <= maxLength + ); + } + + private isCanonicalTimestamp(value: string): boolean { + const timestamp = Date.parse(value); + return ( + Number.isFinite(timestamp) && new Date(timestamp).toISOString() === value + ); + } +} diff --git a/packages/cli/src/commands/channel/runtime.test.ts b/packages/cli/src/commands/channel/runtime.test.ts index 14addb943ea..4ac93f068bd 100644 --- a/packages/cli/src/commands/channel/runtime.test.ts +++ b/packages/cli/src/commands/channel/runtime.test.ts @@ -1,6 +1,7 @@ import { EventEmitter } from 'node:events'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { + daemonObservedContactsPath, daemonSessionRoutesPath, parseConfiguredChannels, registerPermissionRelay, @@ -42,6 +43,16 @@ it('isolates daemon route stores by workspace hash', () => { expect(daemonSessionRoutesPath('/workspace')).not.toBe(sessionsPath()); }); +it('isolates observed contact stores beside daemon routes', () => { + expect(daemonObservedContactsPath('/workspace')).toBe( + '/tmp/qwen/channels/daemon/workspace-hash/observed-contacts.json', + ); + expect(daemonObservedContactsPath('/other')).toBe( + '/tmp/qwen/channels/daemon/other-hash/observed-contacts.json', + ); + expect(daemonObservedContactsPath('/workspace')).not.toBe(sessionsPath()); +}); + describe('parseConfiguredChannels', () => { beforeEach(() => { delete process.env['TOKEN_LITERAL_VALUE']; diff --git a/packages/cli/src/commands/channel/runtime.ts b/packages/cli/src/commands/channel/runtime.ts index 6b4d692dfe9..5908efef90a 100644 --- a/packages/cli/src/commands/channel/runtime.ts +++ b/packages/cli/src/commands/channel/runtime.ts @@ -41,6 +41,16 @@ export function daemonSessionRoutesPath(workspaceCwd: string): string { ); } +export function daemonObservedContactsPath(workspaceCwd: string): string { + return path.join( + Storage.getGlobalQwenDir(), + 'channels', + 'daemon', + hashDaemonWorkspace(workspaceCwd), + 'observed-contacts.json', + ); +} + export function channelLoopPath(): string { return path.join(Storage.getGlobalQwenDir(), 'channels', 'cron.json'); } diff --git a/packages/cli/src/serve/capabilities.ts b/packages/cli/src/serve/capabilities.ts index 62e9318cb77..db06cd832dd 100644 --- a/packages/cli/src/serve/capabilities.ts +++ b/packages/cli/src/serve/capabilities.ts @@ -283,6 +283,8 @@ export const SERVE_CAPABILITY_REGISTRY = { // Runtime GET/PUT/DELETE control for daemon-managed channel selection. // The route exists even when no selection was supplied at daemon boot. channel_control: { since: 'v1' }, + // Read-only workspace graph of recently observed channel contacts. + workspace_channel_observed_contacts: { since: 'v1' }, // Multi-workspace session routing. Advertised only when one daemon hosts // more than one registered workspace runtime. multi_workspace_sessions: { since: 'v1' }, diff --git a/packages/cli/src/serve/routes/workspace-channel-observed-contacts.test.ts b/packages/cli/src/serve/routes/workspace-channel-observed-contacts.test.ts new file mode 100644 index 00000000000..904308e9080 --- /dev/null +++ b/packages/cli/src/serve/routes/workspace-channel-observed-contacts.test.ts @@ -0,0 +1,268 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { promises as fsp } from 'node:fs'; +import * as os from 'node:os'; +import * as path from 'node:path'; +import express from 'express'; +import request from 'supertest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { daemonObservedContactsPath } from '../../commands/channel/runtime.js'; +import { ObservedChannelContactStore } from '../../commands/channel/observed-contact-store.js'; +import { + createWorkspaceRegistry, + type WorkspaceRegistry, + type WorkspaceRuntime, +} from '../workspace-registry.js'; +import { registerWorkspaceChannelObservedContactRoutes } from './workspace-channel-observed-contacts.js'; + +function runtime( + workspaceId: string, + workspaceCwd: string, + trusted = true, +): WorkspaceRuntime { + return { + workspaceId, + workspaceCwd, + primary: workspaceId === 'primary', + trusted, + } as WorkspaceRuntime; +} + +function registry(runtimes: WorkspaceRuntime[]): WorkspaceRegistry { + return createWorkspaceRegistry(runtimes); +} + +describe('workspace observed channel contact routes', () => { + let qwenHome: string; + let previousQwenHome: string | undefined; + + beforeEach(async () => { + previousQwenHome = process.env['QWEN_HOME']; + qwenHome = await fsp.mkdtemp( + path.join(os.tmpdir(), 'qwen-observed-contact-routes-'), + ); + process.env['QWEN_HOME'] = qwenHome; + }); + + afterEach(async () => { + vi.restoreAllMocks(); + if (previousQwenHome === undefined) delete process.env['QWEN_HOME']; + else process.env['QWEN_HOME'] = previousQwenHome; + await fsp.rm(qwenHome, { recursive: true, force: true }); + }); + + it('returns complete direct users and observed group/topic membership', async () => { + const primary = runtime('primary', '/work/main'); + const secondary = runtime('secondary', '/work/secondary'); + new ObservedChannelContactStore( + daemonObservedContactsPath(primary.workspaceCwd), + ).observe('dingtalk-main', { + user: { id: 'direct-primary', label: 'Direct Primary' }, + }); + new ObservedChannelContactStore( + daemonObservedContactsPath(primary.workspaceCwd), + ).observe('dingtalk-main', { + user: { id: 'user-primary', label: 'Primary User' }, + group: { id: 'group-primary', label: 'group-primary' }, + topic: { id: 'topic-primary', label: 'topic-primary' }, + }); + new ObservedChannelContactStore( + daemonObservedContactsPath(secondary.workspaceCwd), + ).observe('dingtalk-main', { + user: { id: 'user-secondary', label: 'Secondary User' }, + }); + const app = express(); + registerWorkspaceChannelObservedContactRoutes(app, { + primaryWorkspace: primary.workspaceCwd, + workspaceRegistry: registry([primary, secondary]), + }); + + const response = await request(app).get( + '/workspace/channel/observed-contacts', + ); + + expect(response.status).toBe(200); + expect(response.headers['cache-control']).toBe('no-store'); + expect(response.body).toEqual({ + users: [ + { + channelName: 'dingtalk-main', + id: 'direct-primary', + label: 'Direct Primary', + lastObservedAt: expect.any(String), + }, + ], + groups: [ + { + channelName: 'dingtalk-main', + id: 'group-primary', + label: 'group-primary', + lastObservedAt: expect.any(String), + users: [ + { + id: 'user-primary', + label: 'Primary User', + lastObservedAt: expect.any(String), + }, + ], + topics: [ + { + id: 'topic-primary', + label: 'topic-primary', + lastObservedAt: expect.any(String), + users: [ + { + id: 'user-primary', + label: 'Primary User', + lastObservedAt: expect.any(String), + }, + ], + }, + ], + }, + ], + }); + expect(JSON.stringify(response.body)).not.toContain('user-secondary'); + }); + + it('selects an exact trusted workspace and returns an empty graph for no file', async () => { + const primary = runtime('primary', '/work/main'); + const secondary = runtime('secondary', '/work/secondary'); + new ObservedChannelContactStore( + daemonObservedContactsPath(secondary.workspaceCwd), + ).observe('telegram-team', { + user: { id: '42', label: 'Ada' }, + }); + const app = express(); + registerWorkspaceChannelObservedContactRoutes(app, { + primaryWorkspace: primary.workspaceCwd, + workspaceRegistry: registry([primary, secondary]), + }); + + const selected = await request(app).get( + '/workspaces/secondary/channel/observed-contacts', + ); + const empty = await request(app).get( + '/workspace/channel/observed-contacts', + ); + + expect(selected.status).toBe(200); + expect(selected.body.users[0]).toMatchObject({ + channelName: 'telegram-team', + label: 'Ada', + id: '42', + }); + expect(empty.body).toEqual({ users: [], groups: [] }); + }); + + it('validates freshness bounds and query shape', async () => { + const primary = runtime('primary', '/work/main'); + const app = express(); + registerWorkspaceChannelObservedContactRoutes(app, { + primaryWorkspace: primary.workspaceCwd, + workspaceRegistry: registry([primary]), + }); + + const valid = await request(app).get( + '/workspace/channel/observed-contacts?freshWithinSeconds=60', + ); + const zero = await request(app).get( + '/workspace/channel/observed-contacts?freshWithinSeconds=0', + ); + const nonNumeric = await request(app).get( + '/workspace/channel/observed-contacts?freshWithinSeconds=recent', + ); + const repeated = await request(app).get( + '/workspace/channel/observed-contacts?freshWithinSeconds=60&freshWithinSeconds=120', + ); + const tooLarge = await request(app).get( + '/workspace/channel/observed-contacts?freshWithinSeconds=31536001', + ); + + expect(valid.status).toBe(200); + for (const invalid of [zero, nonNumeric, repeated, tooLarge]) { + expect(invalid.status).toBe(400); + expect(invalid.body.code).toBe('invalid_freshness'); + } + }); + + it('defaults freshness to seven days', async () => { + const primary = runtime('primary', '/work/main'); + new ObservedChannelContactStore( + daemonObservedContactsPath(primary.workspaceCwd), + { now: () => new Date(Date.now() - 8 * 24 * 60 * 60 * 1000) }, + ).observe('telegram', { + user: { id: 'stale-default-user', label: 'Stale Default User' }, + }); + const app = express(); + registerWorkspaceChannelObservedContactRoutes(app, { + primaryWorkspace: primary.workspaceCwd, + workspaceRegistry: registry([primary]), + }); + + const defaultWindow = await request(app).get( + '/workspace/channel/observed-contacts', + ); + const widerWindow = await request(app).get( + '/workspace/channel/observed-contacts?freshWithinSeconds=777600', + ); + + expect(defaultWindow.body.users).toEqual([]); + expect(widerWindow.body.users[0]?.id).toBe('stale-default-user'); + }); + + it('does not fall back for unknown or untrusted workspace selectors', async () => { + const primary = runtime('primary', '/work/main'); + const untrusted = runtime('untrusted', '/work/untrusted', false); + const app = express(); + registerWorkspaceChannelObservedContactRoutes(app, { + primaryWorkspace: primary.workspaceCwd, + workspaceRegistry: registry([primary, untrusted]), + }); + + const unknown = await request(app).get( + '/workspaces/missing/channel/observed-contacts', + ); + const denied = await request(app).get( + '/workspaces/untrusted/channel/observed-contacts', + ); + + expect(unknown.status).toBe(400); + expect(unknown.body.code).toBe('workspace_mismatch'); + expect(denied.status).toBe(403); + expect(denied.body.code).toBe('untrusted_workspace'); + }); + + it('returns a sanitized error for malformed registry data', async () => { + const primary = runtime('primary', '/work/main'); + const filePath = daemonObservedContactsPath(primary.workspaceCwd); + await fsp.mkdir(path.dirname(filePath), { recursive: true }); + await fsp.writeFile(filePath, '{invalid-json', 'utf8'); + const stderr = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => true); + const app = express(); + registerWorkspaceChannelObservedContactRoutes(app, { + primaryWorkspace: primary.workspaceCwd, + workspaceRegistry: registry([primary]), + }); + + const response = await request(app).get( + '/workspace/channel/observed-contacts', + ); + + expect(response.status).toBe(500); + expect(response.body).toEqual({ + error: 'Observed channel contacts are unavailable.', + code: 'channel_observed_contacts_unavailable', + }); + expect(JSON.stringify(response.body)).not.toContain(filePath); + expect(stderr).toHaveBeenCalledWith( + 'qwen serve: observed channel contacts unavailable.\n', + ); + }); +}); diff --git a/packages/cli/src/serve/routes/workspace-channel-observed-contacts.ts b/packages/cli/src/serve/routes/workspace-channel-observed-contacts.ts new file mode 100644 index 00000000000..234933fbd04 --- /dev/null +++ b/packages/cli/src/serve/routes/workspace-channel-observed-contacts.ts @@ -0,0 +1,85 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { Application, Request, Response } from 'express'; +import { daemonObservedContactsPath } from '../../commands/channel/runtime.js'; +import { + OBSERVED_CONTACT_MAX_FRESH_WITHIN_SECONDS, + ObservedChannelContactStore, +} from '../../commands/channel/observed-contact-store.js'; +import { + requireTrustedWorkspaceRuntime, + resolveWorkspaceRuntimeFromParam, +} from '../workspace-route-runtime.js'; +import type { WorkspaceRegistry } from '../workspace-registry.js'; + +interface RegisterWorkspaceChannelObservedContactRoutesDeps { + primaryWorkspace: string; + workspaceRegistry: WorkspaceRegistry; +} + +const DEFAULT_FRESH_WITHIN_SECONDS = 7 * 24 * 60 * 60; + +function parseFreshWithinSeconds(req: Request): number | undefined { + const raw = req.query['freshWithinSeconds']; + if (raw === undefined) return DEFAULT_FRESH_WITHIN_SECONDS; + if (typeof raw !== 'string' || !/^\d+$/u.test(raw)) return undefined; + const value = Number(raw); + if ( + !Number.isSafeInteger(value) || + value < 1 || + value > OBSERVED_CONTACT_MAX_FRESH_WITHIN_SECONDS + ) { + return undefined; + } + return value; +} + +function sendContacts(req: Request, res: Response, workspaceCwd: string): void { + res.setHeader('Cache-Control', 'no-store'); + res.setHeader('X-Content-Type-Options', 'nosniff'); + const freshWithinSeconds = parseFreshWithinSeconds(req); + if (freshWithinSeconds === undefined) { + res.status(400).json({ + error: `freshWithinSeconds must be an integer from 1 to ${OBSERVED_CONTACT_MAX_FRESH_WITHIN_SECONDS}.`, + code: 'invalid_freshness', + }); + return; + } + try { + const store = new ObservedChannelContactStore( + daemonObservedContactsPath(workspaceCwd), + ); + res.status(200).json(store.list({ freshWithinSeconds })); + } catch { + process.stderr.write( + 'qwen serve: observed channel contacts unavailable.\n', + ); + res.status(500).json({ + error: 'Observed channel contacts are unavailable.', + code: 'channel_observed_contacts_unavailable', + }); + } +} + +export function registerWorkspaceChannelObservedContactRoutes( + app: Application, + deps: RegisterWorkspaceChannelObservedContactRoutesDeps, +): void { + app.get('/workspace/channel/observed-contacts', (req, res) => { + sendContacts(req, res, deps.primaryWorkspace); + }); + + app.get('/workspaces/:workspace/channel/observed-contacts', (req, res) => { + const runtime = resolveWorkspaceRuntimeFromParam( + deps.workspaceRegistry, + req, + res, + ); + if (!runtime || !requireTrustedWorkspaceRuntime(runtime, res)) return; + sendContacts(req, res, runtime.workspaceCwd); + }); +} diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index c2e3c35a15a..0e89dc05d3d 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -324,6 +324,7 @@ const EXPECTED_STAGE1_FEATURES = [ 'session_hooks', 'workspace_extensions', 'session_branch', + 'workspace_channel_observed_contacts', 'workspace_qualified_rest_core', 'extension_management_v2', 'workspace_persisted_transcript', @@ -374,6 +375,7 @@ const EXPECTED_REGISTERED_FEATURES = [ f !== 'session_hooks' && f !== 'workspace_extensions' && f !== 'session_branch' && + f !== 'workspace_channel_observed_contacts' && f !== 'workspace_qualified_rest_core' && f !== 'extension_management_v2' && f !== 'workspace_persisted_transcript' && @@ -412,6 +414,7 @@ const EXPECTED_REGISTERED_FEATURES = [ 'workspace_reload', 'channel_reload', 'channel_control', + 'workspace_channel_observed_contacts', 'multi_workspace_sessions', 'multi_workspace_session_rewind', 'multi_workspace_session_shell', @@ -2877,6 +2880,40 @@ describe('createServeApp', () => { }); }); + describe('GET /workspace/channel/observed-contacts', () => { + it('is assembled behind bearer authentication', async () => { + const previousQwenHome = process.env['QWEN_HOME']; + const qwenHome = await fsp.mkdtemp( + path.join(os.tmpdir(), 'qwen-observed-contact-server-'), + ); + process.env['QWEN_HOME'] = qwenHome; + try { + const app = createServeApp({ + ...baseOpts, + token: 'secret', + workspace: WS_BOUND, + }); + const host = `127.0.0.1:${baseOpts.port}`; + + const unauthenticated = await request(app) + .get('/workspace/channel/observed-contacts') + .set('Host', host); + const authenticated = await request(app) + .get('/workspace/channel/observed-contacts') + .set('Host', host) + .set('Authorization', 'Bearer secret'); + + expect(unauthenticated.status).toBe(401); + expect(authenticated.status).toBe(200); + expect(authenticated.body).toEqual({ users: [], groups: [] }); + } finally { + if (previousQwenHome === undefined) delete process.env['QWEN_HOME']; + else process.env['QWEN_HOME'] = previousQwenHome; + await fsp.rm(qwenHome, { recursive: true, force: true }); + } + }); + }); + describe('GET /capabilities', () => { it('advertises session generation only when every bridge supports it', async () => { const supported = await request( diff --git a/packages/cli/src/serve/server.ts b/packages/cli/src/serve/server.ts index bea17fa53ac..2bbf5e51068 100644 --- a/packages/cli/src/serve/server.ts +++ b/packages/cli/src/serve/server.ts @@ -196,6 +196,7 @@ import { registerWorkspaceQualifiedMcpControlRoutes, } from './routes/workspace-mcp-control.js'; import { registerWorkspaceChannelControlRoutes } from './routes/workspace-channel-control.js'; +import { registerWorkspaceChannelObservedContactRoutes } from './routes/workspace-channel-observed-contacts.js'; import { registerWorkspaceQualifiedToolsRoutes, registerWorkspaceToolsRoutes, @@ -1486,6 +1487,10 @@ export function createServeApp( parseAndValidateWorkspaceClientId(req, res, primaryBridge), }); } + registerWorkspaceChannelObservedContactRoutes(app, { + primaryWorkspace: primaryBoundWorkspace, + workspaceRegistry, + }); registerWorkspaceLifecycleRoutes(app, { boundWorkspace: primaryBoundWorkspace, workspace: primaryWorkspace,