fix: support Harness interactions in DingTalk and Feishu

This commit is contained in:
xmanrui 2026-08-18 13:03:15 +08:00
parent c9ec84cc67
commit 3e7e1dcdd7
13 changed files with 5061 additions and 881 deletions

File diff suppressed because one or more lines are too long

View file

@ -3,11 +3,17 @@ import {
splitDingtalkText,
} from './dingtalk-api.mjs';
import { createDingTalkCardStream } from './dingtalk-card-stream.mjs';
import {
harnessAnswerForQuestion,
harnessQuestionText,
validHarnessQuestion,
} from '../shared/harness-question.mjs';
import { runWorkspaceCommand } from '../shared/workspace-command.mjs';
import { askInWorkspaceSession } from '../shared/workspace-session.mjs';
const CARD_INITIAL_TEXT = '已连接 DeepSeek Harness,正在思考…';
const CARD_ERROR_TEXT = '消息处理失败,请稍后重试。';
const INTERACTION_RESOLVED_TEXT = '这个问题已在其他客户端处理,无需再次回答。';
const HELP_TEXT = [
'钉钉机器人已连接 DeepSeek Harness。',
@ -55,6 +61,19 @@ function progressText(update) {
return `_${nonEmptyString(update?.text) ?? '正在处理…'}_`;
}
function canClaimInteractionReply(message, pending, sender) {
if (pending.needsPresentation || !pending.questions[pending.index]) return false;
if (pending.actor !== sender) return false;
if (String(message?.conversationType) === '2' && message?.isInAtList !== true) return false;
if (message?.msgtype !== 'text' || !nonEmptyString(message?.text?.content)) return false;
try {
normalizeDingtalkSessionWebhook(message.sessionWebhook);
return true;
} catch {
return false;
}
}
function ensureStats(status) {
status.stats ??= {};
for (const key of ['messagesReceived', 'messagesReplied', 'messagesRejected', 'messagesIgnored']) {
@ -102,6 +121,8 @@ export class DingtalkHarnessBridge {
#maxMessageChars;
#signal;
#queues = new Map();
#pendingInteractions = new Map();
#interactionKeys = new Map();
#acceptedMessageIds = new Set();
constructor({
@ -157,12 +178,49 @@ export class DingtalkHarnessBridge {
this.#status.lastRejectedAt = new Date().toISOString();
return Promise.resolve();
}
const pending = this.#pendingInteractions.get(key);
if (pending && pending.actor !== sender) {
return this.#enqueueMessage(message, messageId, sender, key);
}
// Once one valid answer has been claimed, later messages are subsequent
// prompts even if the network submission eventually needs a retry. Invalid
// replies do not claim the question, so the next valid answer can still
// pass through this interaction queue.
if (pending?.submitting || pending?.claimedReplyMessageId) {
return this.#enqueueMessage(message, messageId, sender, key);
}
if (pending) {
if (canClaimInteractionReply(message, pending, sender)) {
pending.claimedReplyMessageId = messageId;
}
const previous = pending.queue ?? Promise.resolve();
const current = previous
.catch(() => undefined)
.then(() => this.#processInteractionReply(message, messageId, sender, key, pending))
.finally(() => {
this.#acceptedMessageIds.delete(messageId);
if (pending.claimedReplyMessageId === messageId) {
pending.claimedReplyMessageId = null;
}
if (pending.queue === current) pending.queue = null;
});
pending.queue = current;
return current;
}
return this.#enqueueMessage(message, messageId, sender, key);
}
#enqueueMessage(message, messageId, sender, key, {
releaseMessageId = true,
alreadyRecorded = false,
} = {}) {
const previous = this.#queues.get(key) ?? Promise.resolve();
const current = previous
.catch(() => undefined)
.then(() => this.#process(message, messageId, sender, key))
.then(() => this.#process(message, messageId, sender, key, { alreadyRecorded }))
.finally(() => {
this.#acceptedMessageIds.delete(messageId);
if (releaseMessageId) this.#acceptedMessageIds.delete(messageId);
if (this.#queues.get(key) === current) this.#queues.delete(key);
});
this.#queues.set(key, current);
@ -170,15 +228,22 @@ export class DingtalkHarnessBridge {
}
async waitForIdle() {
await Promise.allSettled([...this.#queues.values()]);
await Promise.allSettled([
...this.#queues.values(),
...[...this.#pendingInteractions.values()].flatMap((pending) => (
pending.queue ? [pending.queue] : []
)),
]);
}
async #process(message, messageId, sender, key) {
async #process(message, messageId, sender, key, { alreadyRecorded = false } = {}) {
this.#signal?.throwIfAborted();
if (this.#state.hasSeen(messageId)) return;
await this.#state.markSeen(messageId);
increment(this.#status, 'messagesReceived');
this.#status.lastMessageAt = new Date().toISOString();
if (!alreadyRecorded) {
if (this.#state.hasSeen(messageId)) return;
await this.#state.markSeen(messageId);
increment(this.#status, 'messagesReceived');
this.#status.lastMessageAt = new Date().toISOString();
}
if (String(message.conversationType) === '2' && message.isInAtList !== true) {
increment(this.#status, 'messagesIgnored');
@ -253,6 +318,13 @@ export class DingtalkHarnessBridge {
onUpdate: cardStarted
? (update) => cardStream.push(progressText(update))
: undefined,
onInteraction: (interaction) => this.#handleInteraction(interaction, {
key,
actor: sender,
sessionWebhook,
requiresMention: String(message.conversationType) === '2',
}),
onInteractionResolved: (resolution) => this.#handleInteractionResolved(resolution),
},
});
const streamed = cardStarted && await cardStream.finish(answer);
@ -270,6 +342,286 @@ export class DingtalkHarnessBridge {
} catch {
this.#logger.error?.('[dsh-dingtalk] failed to send the safe error reply');
}
} finally {
await this.#cancelPendingInteraction(key);
}
}
async #processInteractionReply(message, messageId, sender, key, expected) {
this.#signal?.throwIfAborted();
const current = this.#pendingInteractions.get(key);
const claimed = expected.claimedReplyMessageId === messageId;
if (!current || current !== expected || current.submitting) {
if (claimed && (!current || current !== expected)) {
return this.#discardResolvedInteractionReply(message, messageId);
}
return this.#enqueueMessage(message, messageId, sender, key, { releaseMessageId: false });
}
if (this.#state.hasSeen(messageId)) return;
await this.#state.markSeen(messageId);
increment(this.#status, 'messagesReceived');
this.#status.lastMessageAt = new Date().toISOString();
if (String(message.conversationType) === '2' && message.isInAtList !== true) {
increment(this.#status, 'messagesIgnored');
return;
}
let sessionWebhook;
try {
sessionWebhook = normalizeDingtalkSessionWebhook(message.sessionWebhook);
} catch {
increment(this.#status, 'messagesRejected');
this.#status.lastRejectedAt = new Date().toISOString();
this.#status.lastError = '钉钉消息没有安全的回复地址。';
return;
}
const text = message?.msgtype === 'text' ? nonEmptyString(message?.text?.content) : null;
if (!text) {
try {
await this.#send(sessionWebhook, '请用文字回答当前问题。');
} catch {
this.#logger.error?.('[dsh-dingtalk] failed to reject a non-text interaction reply');
}
return;
}
const pending = this.#pendingInteractions.get(key);
if (!pending || pending !== expected || pending.submitting) {
if (claimed && (!pending || pending !== expected)) {
try {
await this.#send(sessionWebhook, INTERACTION_RESOLVED_TEXT);
} catch {
this.#logger.error?.('[dsh-dingtalk] failed to send an expired interaction notice');
}
return;
}
return this.#enqueueMessage(message, messageId, sender, key, {
releaseMessageId: false,
alreadyRecorded: true,
});
}
pending.sessionWebhook = sessionWebhook;
if (pending.needsPresentation) {
try {
await this.#presentInteraction(pending);
} catch {
this.#status.lastError = '钉钉交互问题发送失败。';
this.#logger.error?.('[dsh-dingtalk] failed to retry an interaction question');
pending.interaction.reconnect?.();
}
return;
}
const question = pending.questions[pending.index];
if (!question) return;
pending.answers.push(harnessAnswerForQuestion(question, text));
pending.index += 1;
if (pending.index < pending.questions.length) {
if (pending.claimedReplyMessageId === messageId) {
pending.claimedReplyMessageId = null;
}
pending.needsPresentation = true;
try {
await this.#presentInteraction(pending);
} catch {
this.#status.lastError = '钉钉交互问题发送失败。';
this.#logger.error?.('[dsh-dingtalk] failed to send the next interaction question');
pending.interaction.reconnect?.();
}
return;
}
pending.submitting = true;
try {
await pending.interaction.respond({
ok: true,
value: {
sessionId: pending.sessionId,
answer: { answers: pending.answers },
},
});
this.#clearPendingInteraction(key, pending.interactionId);
this.#status.lastError = null;
} catch (error) {
if (this.#signal?.aborted) return;
if (this.#pendingInteractions.get(key) !== pending) return;
if (error?.code === 'interaction-not-pending') {
this.#clearPendingInteraction(key, pending.interactionId);
try {
await this.#send(sessionWebhook, INTERACTION_RESOLVED_TEXT);
} catch {
this.#logger.error?.('[dsh-dingtalk] failed to send an expired interaction notice');
}
return;
}
pending.submitting = false;
pending.answers.pop();
pending.index -= 1;
this.#status.lastError = '回答提交失败。';
this.#logger.error?.('[dsh-dingtalk] failed to answer a Harness interaction');
try {
await this.#send(sessionWebhook, '回答提交失败,请重新发送当前问题的答案。');
} catch {
this.#logger.error?.('[dsh-dingtalk] failed to send an interaction retry notice');
}
}
}
async #handleInteraction(interaction, {
key,
actor,
sessionWebhook,
requiresMention,
}) {
// The transport deliberately exposes every interaction kind. Approval is
// left unanswered until #5 adds its own policy and renderer.
if (interaction?.kind !== 'question') return;
const questions = interaction?.payload?.questions;
const interactionId = typeof interaction?.interactionId === 'string'
? interaction.interactionId
: interaction?.rpcId;
if (typeof interaction.rpcId !== 'string'
|| typeof interactionId !== 'string'
|| typeof interaction.sessionId !== 'string'
|| !Array.isArray(questions)
|| questions.length === 0
|| questions.some((question) => !validHarnessQuestion(question))) {
this.#logger.warn?.('[dsh-dingtalk] ignored an invalid Harness question interaction');
return;
}
if (interaction.recovered === true) {
await interaction.respond({
ok: false,
error: {
code: 'cancelled',
message: 'DingTalk safely cancelled an interaction left by an earlier client.',
details: {},
},
});
try {
await this.#send(sessionWebhook, '检测到这个 Session 中遗留的待回答问题,已安全取消并继续处理你刚才的消息。');
} catch {
this.#logger.error?.('[dsh-dingtalk] failed to send an interaction recovery notice');
}
return;
}
const existing = this.#pendingInteractions.get(key);
if (existing?.interactionId === interactionId) {
existing.interaction = interaction;
if (existing.needsPresentation) await this.#presentInteraction(existing);
return;
}
if (this.#interactionKeys.has(interactionId)) return;
if (existing) {
this.#logger.warn?.('[dsh-dingtalk] cancelled a second pending Harness question');
await interaction.respond({
ok: false,
error: {
code: 'cancelled',
message: 'DingTalk is already handling another user interaction.',
details: {},
},
});
return;
}
const pending = {
kind: 'question',
interactionId,
sessionId: interaction.sessionId,
interaction,
actor,
requiresMention,
questions,
answers: [],
index: 0,
sessionWebhook,
queue: null,
claimedReplyMessageId: null,
submitting: false,
needsPresentation: true,
};
this.#pendingInteractions.set(key, pending);
this.#interactionKeys.set(pending.interactionId, key);
await this.#presentInteraction(pending);
}
#handleInteractionResolved(resolution) {
const interactionId = resolution?.interactionId;
if (resolution?.kind !== 'question' || typeof interactionId !== 'string') return;
const key = this.#interactionKeys.get(interactionId);
if (!key) return;
this.#clearPendingInteraction(key, interactionId);
}
async #presentInteraction(pending) {
const question = pending.questions[pending.index];
if (!question) return;
await this.#send(
pending.sessionWebhook,
harnessQuestionText(
question,
pending.index,
pending.questions.length,
{ requiresMention: pending.requiresMention },
),
);
pending.needsPresentation = false;
}
async #discardResolvedInteractionReply(message, messageId) {
if (this.#state.hasSeen(messageId)) return;
await this.#state.markSeen(messageId);
increment(this.#status, 'messagesReceived');
this.#status.lastMessageAt = new Date().toISOString();
let sessionWebhook;
try {
sessionWebhook = normalizeDingtalkSessionWebhook(message.sessionWebhook);
} catch {
increment(this.#status, 'messagesRejected');
this.#status.lastRejectedAt = new Date().toISOString();
return;
}
try {
await this.#send(sessionWebhook, INTERACTION_RESOLVED_TEXT);
} catch {
this.#logger.error?.('[dsh-dingtalk] failed to send an expired interaction notice');
}
}
#takePendingInteraction(key, interactionId) {
const pending = this.#pendingInteractions.get(key);
if (!pending
|| (interactionId !== undefined && pending.interactionId !== interactionId)) return null;
this.#pendingInteractions.delete(key);
this.#interactionKeys.delete(pending.interactionId);
return pending;
}
#clearPendingInteraction(key, interactionId) {
return this.#takePendingInteraction(key, interactionId) !== null;
}
async #cancelPendingInteraction(key) {
const pending = this.#takePendingInteraction(key);
if (!pending || pending.kind !== 'question') return;
try {
await pending.interaction.respond({
ok: false,
error: {
code: 'cancelled',
message: 'The DingTalk interaction ended before the user answered.',
details: {},
},
}, { signal: AbortSignal.timeout(5_000) });
} catch (error) {
if (error?.code !== 'interaction-not-pending') {
this.#logger.warn?.('[dsh-dingtalk] failed to cancel a pending Harness interaction');
}
}
}

View file

@ -1,369 +1,19 @@
import { spawn } from 'node:child_process';
import { randomUUID } from 'node:crypto';
import { isAbsolute } from 'node:path';
import {
HarnessClient as SharedHarnessClient,
} from '../shared/harness-client.mjs';
import { adoptRegisteredWorkspaceSession } from '../shared/harness-session-binding.mjs';
export {
HarnessInteractionError,
HarnessReplyTracker,
HarnessRpcError,
} from '../shared/harness-client.mjs';
function workspacePaths(value) {
if (!Array.isArray(value?.items)) return [];
return value.items.flatMap((item) => (
typeof item?.path === 'string' && isAbsolute(item.path) ? [item.path] : []
));
}
function workspaceFromList(workspacePath, workspaceList) {
if (!Array.isArray(workspaceList?.items)
|| !Array.isArray(workspaceList?.archivedSessionIds)) {
throw new Error('Harness returned an invalid response for workspace.list');
}
const workspace = workspaceList.items.find((item) => item?.path === workspacePath);
if (!workspace) return null;
if (!Array.isArray(workspace.sessionIds)
|| workspace.sessionIds.some((sessionId) => typeof sessionId !== 'string')) {
throw new Error('Harness returned invalid session IDs for workspace.list');
}
return workspace;
}
function workspaceSessions(workspace, archivedSessionIds, sessionList) {
if (!Array.isArray(sessionList?.items)) {
throw new Error('Harness returned an invalid response for session.list');
}
const archived = new Set(archivedSessionIds);
const summaries = new Map(sessionList.items.flatMap((item) => (
typeof item?.sessionId === 'string' ? [[item.sessionId, item]] : []
)));
return {
workspace: workspace.path,
sessions: workspace.sessionIds.map((sessionId) => {
const summary = summaries.get(sessionId);
const title = summary?.projections?.values?.title;
return {
sessionId,
title: typeof title === 'string' ? title : null,
archived: archived.has(sessionId),
blank: summary?.blank === true,
origin: summary?.origin === 'subagent' ? 'subagent' : null,
summaryAvailable: summary !== undefined,
};
}),
};
}
function sleep(ms, signal) {
return new Promise((resolve, reject) => {
if (signal?.aborted) {
reject(signal.reason ?? new DOMException('Aborted', 'AbortError'));
return;
}
const timer = setTimeout(() => {
signal?.removeEventListener('abort', onAbort);
resolve();
}, ms);
const onAbort = () => {
clearTimeout(timer);
reject(signal.reason ?? new DOMException('Aborted', 'AbortError'));
};
signal?.addEventListener('abort', onAbort, { once: true });
});
}
function assistantMessageText(event) {
return (event?.data?.message?.content ?? [])
.filter((part) => part.type === 'text' && typeof part.text === 'string')
.map((part) => part.text)
.join('\n')
.trim();
}
export class HarnessReplyTracker {
#promptRpcId;
#lastSeq;
#openTurn = null;
#targetTurn = null;
#stepText = new Map();
#latestText = '';
#finished = false;
#reason = null;
constructor({ promptRpcId, afterSeq = -1 }) {
this.#promptRpcId = promptRpcId;
this.#lastSeq = afterSeq;
}
get finished() {
return this.#finished;
}
get answer() {
return this.#latestText.trim();
}
get reason() {
return this.#reason;
}
consume(entries) {
let update = null;
const ordered = [...entries]
.map((entry) => entry?.event ?? entry)
.filter(Boolean)
.sort((left, right) => (left.seq ?? -1) - (right.seq ?? -1));
for (const event of ordered) {
const seq = event.seq ?? -1;
if (seq <= this.#lastSeq) continue;
this.#lastSeq = seq;
if (event.type === 'turn/start') this.#openTurn = event.data?.turn ?? null;
if (event.type === 'user/message' && event.data?.source?.rpcId === this.#promptRpcId) {
this.#targetTurn = this.#openTurn;
continue;
}
if (this.#targetTurn === null) continue;
if (event.type === 'turn/end') {
if (event.data?.turn !== this.#targetTurn) continue;
this.#finished = true;
this.#reason = event.data?.reason ?? null;
this.#openTurn = null;
continue;
}
if (event.data?.turn !== this.#targetTurn) continue;
if (event.type === 'assistant/chunk' && event.data?.chunk?.type === 'text-delta') {
const step = event.data?.step ?? 0;
const index = event.data.chunk.index ?? 0;
const key = `${step}:${index}`;
this.#stepText.set(key, (this.#stepText.get(key) ?? '') + event.data.chunk.text);
const prefix = `${step}:`;
const text = [...this.#stepText.entries()]
.filter(([partKey]) => partKey.startsWith(prefix))
.sort(([left], [right]) => Number(left.split(':')[1]) - Number(right.split(':')[1]))
.map(([, part]) => part)
.join('\n')
.trim();
if (text && text !== this.#latestText) {
this.#latestText = text;
update = { type: 'text', text };
}
continue;
}
if (event.type === 'assistant/message') {
const text = assistantMessageText(event);
if (text && text !== this.#latestText) {
this.#latestText = text;
update = { type: 'text', text };
}
continue;
}
if (event.type === 'tool/call') {
update = { type: 'tool', name: event.data?.name ?? '工具' };
} else if (event.type === 'tool/result') {
update = { type: 'status', text: '正在整理结果…' };
}
}
return update;
}
}
export class HarnessRpcError extends Error {
constructor(method, error) {
super(`${method}: ${error?.message ?? 'unknown Harness RPC error'}`);
this.name = 'HarnessRpcError';
this.method = method;
this.code = error?.code ?? 'internal';
this.details = error?.details ?? {};
}
}
export class HarnessClient {
#baseUrl;
#workspace;
#agentPreset;
#autostart;
#dshBin;
#fetch;
#managedProcess = null;
constructor({
baseUrl,
workspace,
agentPreset = 'standard',
autostart = false,
dshBin = 'dsh',
fetchImpl = fetch,
}) {
this.#baseUrl = new URL(baseUrl);
this.#workspace = workspace;
this.#agentPreset = agentPreset;
this.#autostart = autostart;
this.#dshBin = dshBin;
this.#fetch = fetchImpl;
}
async rpc(method, payload = {}, timeoutMs = 30_000, options = {}) {
const rpcId = options.rpcId ?? `dingtalk-${randomUUID()}`;
const timeoutSignal = AbortSignal.timeout(timeoutMs);
const signal = options.signal
? AbortSignal.any([options.signal, timeoutSignal])
: timeoutSignal;
const response = await this.#fetch(new URL(`/api/${method}`, this.#baseUrl), {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify({ type: 'client-request', rpcId, method, payload }),
signal,
export class HarnessClient extends SharedHarnessClient {
constructor(options) {
super({
...options,
rpcIdPrefix: 'dingtalk',
logPrefix: 'dsh-dingtalk',
});
if (!response.ok) throw new Error(`Harness transport ${method} failed: HTTP ${response.status}`);
const body = await response.json();
if (body?.type !== 'server-response' || body?.rpcId !== rpcId) {
throw new Error(`Harness returned an invalid response for ${method}`);
}
if (!body.result?.ok) throw new HarnessRpcError(method, body.result?.error);
return body.result.value;
}
async health(options = {}) {
await this.rpc('host.describe', {}, 5_000, options);
return true;
}
async ensureRunning(options = {}) {
try {
return await this.health(options);
} catch (firstError) {
if (!this.#autostart) throw firstError;
}
if (!this.#managedProcess || this.#managedProcess.exitCode !== null) {
const port = this.#baseUrl.port || (this.#baseUrl.protocol === 'https:' ? '443' : '80');
this.#managedProcess = spawn(this.#dshBin, [
'web', '--host', this.#baseUrl.hostname, '--port', port,
], {
cwd: this.#workspace,
env: process.env,
stdio: ['ignore', 'inherit', 'inherit'],
});
this.#managedProcess.on('error', (error) => {
console.error('[dsh-dingtalk] failed to start Harness:', error.message);
});
}
const deadline = Date.now() + 60_000;
let lastError;
while (Date.now() < deadline) {
await sleep(1_000, options.signal);
try {
return await this.health(options);
} catch (error) {
lastError = error;
}
}
throw new Error(`Harness did not become ready: ${lastError?.message ?? 'timeout'}`);
}
async listWorkspaces(options = {}) {
await this.ensureRunning(options);
return workspacePaths(await this.rpc('workspace.list', {}, 30_000, options));
}
async listWorkspaceSessions(workspacePath, options = {}) {
await this.ensureRunning(options);
const workspaceList = await this.rpc('workspace.list', {}, 30_000, options);
const workspace = workspaceFromList(workspacePath, workspaceList);
if (!workspace) return { workspace: workspacePath, sessions: [] };
const sessionList = await this.rpc('session.list', {}, 30_000, options);
return workspaceSessions(workspace, workspaceList.archivedSessionIds, sessionList);
}
async adoptWorkspaceSession(value, options = {}) {
return adoptRegisteredWorkspaceSession(this, value, options);
}
async workspaceId(options = {}) {
const { workspace = this.#workspace, ...rpcOptions } = options;
const { items } = await this.rpc('workspace.list', {}, 30_000, rpcOptions);
const existing = items.find((item) => item.path === workspace);
if (existing) return existing.workspaceId;
const created = await this.rpc('workspace.create', { path: workspace }, 30_000, rpcOptions);
return created.workspace.workspaceId;
}
async createSession(options = {}) {
await this.ensureRunning(options);
const workspaceId = await this.workspaceId(options);
const created = await this.rpc('session.create', {
workspaceId,
agentPreset: this.#agentPreset,
}, 30_000, options);
return created.sessionId;
}
async sessionExists(sessionId, options = {}) {
try {
await this.rpc('session.history', { sessionId, maxMessages: 1 }, 30_000, options);
return true;
} catch (error) {
if (error instanceof HarnessRpcError && error.code === 'session-not-found') return false;
throw error;
}
}
async ask(sessionId, text, options = {}) {
if (typeof options === 'number') options = { timeoutMs: options };
const timeoutMs = options.timeoutMs ?? 600_000;
const signal = options.signal;
const onUpdate = typeof options.onUpdate === 'function' ? options.onUpdate : null;
await this.ensureRunning({ signal });
const before = await this.rpc(
'session.history',
{ sessionId, maxMessages: 1 },
30_000,
{ signal },
);
const baselineSeq = Math.max(-1, ...(before.events ?? []).map(({ event }) => event.seq ?? -1));
const promptRpcId = `dingtalk-${randomUUID()}`;
const tracker = new HarnessReplyTracker({ promptRpcId, afterSeq: baselineSeq });
await this.rpc('session.prompt', {
sessionId,
mode: 'queue',
content: [{ type: 'text', text }],
clientTimeZone: Intl.DateTimeFormat().resolvedOptions().timeZone,
}, 30_000, { rpcId: promptRpcId, signal });
const deadline = Date.now() + timeoutMs;
while (Date.now() < deadline) {
await sleep(300, signal);
const history = await this.rpc(
'session.history',
{ sessionId, maxMessages: 50 },
30_000,
{ signal },
);
const update = tracker.consume(history.events ?? []);
if (update && onUpdate) {
try {
await onUpdate(update);
} catch (error) {
console.warn('[dsh-dingtalk] ignored a progress update failure:', error.message);
}
}
if (!tracker.finished) continue;
if (tracker.answer) return tracker.answer;
throw new Error(
`Harness turn ended without a text reply${tracker.reason ? ` (${JSON.stringify(tracker.reason)})` : ''}`,
);
}
throw new Error(`Harness reply timed out after ${Math.round(timeoutMs / 1_000)} seconds`);
}
stopManagedProcess() {
if (this.#managedProcess?.exitCode === null) this.#managedProcess.kill('SIGTERM');
}
}

View file

@ -5,9 +5,17 @@ import {
isBotSender,
splitText,
} from './message-utils.mjs';
import {
harnessAnswerForQuestion,
harnessQuestionText,
validHarnessQuestion,
} from '../shared/harness-question.mjs';
import { runWorkspaceCommand } from '../shared/workspace-command.mjs';
import { askInWorkspaceSession } from '../shared/workspace-session.mjs';
const INTERACTION_RESOLVED_TEXT = '这个问题已在其他客户端处理,无需再次回答。';
const RESOLVED_REPLY_TTL_MS = 30 * 60_000;
const HELP_TEXT = [
'北汇星河 AIOS 已连接 DeepSeek Harness。',
'',
@ -21,16 +29,49 @@ const HELP_TEXT = [
'/help 显示本帮助',
].join('\n');
function nonEmptyString(value) {
return typeof value === 'string' && value.trim() ? value.trim() : null;
}
function senderOpenId(event) {
return nonEmptyString(event?.sender?.sender_id?.open_id)
?? nonEmptyString(event?.sender?.sender_id?.user_id);
}
function canClaimInteractionReply(event, pending) {
return pending.needsPresentation !== true
&& pending.questions[pending.index]
&& senderOpenId(event) === pending.actor
&& event?.message?.message_type === 'text'
&& nonEmptyString(extractText(event));
}
function ensureStatus(status) {
for (const key of ['messagesReceived', 'messagesReplied', 'messagesRejected']) {
status[key] ??= 0;
}
status.lastMessageAt ??= null;
status.lastReplyAt ??= null;
status.lastRejectedAt ??= null;
status.lastError ??= null;
}
export class FeishuHarnessBridge {
#client;
#channel;
#harness;
#state;
#queues = new Map();
#pendingInteractions = new Map();
#interactionKeys = new Map();
#resolvedQuestionReplies = new Map();
#acceptedMessageIds = new Set();
#interactionTasks = new Set();
#status;
#allowedSenderOpenIds;
#replyTimeoutMs;
#logger;
#signal;
constructor({
client,
@ -39,8 +80,13 @@ export class FeishuHarnessBridge {
state,
status,
allowedSenderOpenIds = new Set(),
replyTimeoutMs = 600000,
replyTimeoutMs = 600_000,
logger = console,
signal,
}) {
if (!client || !harness || !state || !status) {
throw new TypeError('Feishu bridge dependencies are required');
}
this.#client = client;
this.#channel = channel;
this.#harness = harness;
@ -48,55 +94,171 @@ export class FeishuHarnessBridge {
this.#status = status;
this.#allowedSenderOpenIds = allowedSenderOpenIds;
this.#replyTimeoutMs = replyTimeoutMs;
this.#logger = logger;
this.#signal = signal;
ensureStatus(this.#status);
}
accept(event) {
const messageId = event?.message?.message_id;
if (!messageId || isBotSender(event) || event?.message?.message_type !== 'text') return;
if (this.#signal?.aborted) return Promise.resolve();
const messageId = nonEmptyString(event?.message?.message_id);
if (!messageId || isBotSender(event)) return Promise.resolve();
if (!isAllowedSender(event, this.#allowedSenderOpenIds)) {
this.#status.messagesRejected += 1;
this.#status.lastRejectedAt = new Date().toISOString();
console.warn('[bridge] ignored a message from a sender outside the allowlist');
return;
this.#logger.warn?.('[dsh-feishu] ignored a message from a sender outside the allowlist');
return Promise.resolve();
}
if (this.#state.hasSeen(messageId) || this.#acceptedMessageIds.has(messageId)) return;
if (this.#state.hasSeen(messageId) || this.#acceptedMessageIds.has(messageId)) {
return Promise.resolve();
}
let key;
try {
key = conversationKey(event);
} catch {
this.#status.messagesRejected += 1;
this.#status.lastRejectedAt = new Date().toISOString();
return Promise.resolve();
}
this.#acceptedMessageIds.add(messageId);
const processingReaction = this.#addReaction(messageId, 'OnIt');
if (this.#isResolvedQuestionReply(event, key)) {
const current = Promise.resolve()
.then(() => this.#discardResolvedInteractionReply(event, messageId))
.then(() => this.#finishReaction(messageId, processingReaction, 'DONE'))
.catch((error) => this.#handleMessageFailure(
event,
messageId,
processingReaction,
error,
))
.finally(() => this.#acceptedMessageIds.delete(messageId));
return current;
}
const pending = this.#pendingInteractions.get(key);
if (pending && senderOpenId(event) !== pending.actor) {
return this.#enqueueMessage(event, messageId, key, processingReaction);
}
if (pending?.submitting || pending?.claimedReplyMessageId) {
return this.#enqueueMessage(event, messageId, key, processingReaction);
}
if (pending) {
if (canClaimInteractionReply(event, pending)) pending.claimedReplyMessageId = messageId;
const previous = pending.queue ?? Promise.resolve();
const processing = previous
.catch(() => undefined)
.then(() => this.#processInteractionReply(
event,
messageId,
key,
pending,
processingReaction,
));
pending.queue = processing;
const key = conversationKey(event);
const releaseInteraction = () => {
if (pending.claimedReplyMessageId === messageId) {
pending.claimedReplyMessageId = null;
}
if (pending.queue === processing) pending.queue = null;
};
let current;
current = processing
.then(
() => {
releaseInteraction();
return this.#finishReaction(messageId, processingReaction, 'DONE');
},
(error) => {
releaseInteraction();
return this.#handleMessageFailure(
event,
messageId,
processingReaction,
error,
);
},
)
.finally(() => {
releaseInteraction();
this.#acceptedMessageIds.delete(messageId);
this.#interactionTasks.delete(current);
});
this.#interactionTasks.add(current);
return current;
}
return this.#enqueueMessage(event, messageId, key, processingReaction);
}
#enqueueMessage(event, messageId, key, processingReaction, {
releaseMessageId = true,
alreadyRecorded = false,
finalize = true,
} = {}) {
const previous = this.#queues.get(key) ?? Promise.resolve();
const task = previous
const work = previous
.catch(() => undefined)
.then(() => this.#handle(event, key))
.then(() => this.#finishReaction(messageId, processingReaction, 'DONE'))
.catch(async (error) => {
console.error('[bridge] message handling failed:', error.message);
this.#status.lastError = error.message;
await this.#finishReaction(messageId, processingReaction, 'ERROR');
await this.#send(
event.message.chat_id,
'处理失败,请稍后重试。如果问题持续,请在 DeepSeek Harness 的飞书插件页面检查连接状态。',
).catch(() => undefined);
})
.finally(() => {
this.#acceptedMessageIds.delete(messageId);
if (this.#queues.get(key) === task) this.#queues.delete(key);
});
this.#queues.set(key, task);
.then(() => this.#handle(event, key, { alreadyRecorded }));
const settled = finalize
? work
.then(() => this.#finishReaction(messageId, processingReaction, 'DONE'))
.catch((error) => this.#handleMessageFailure(
event,
messageId,
processingReaction,
error,
))
: work;
let current;
current = settled.finally(() => {
if (releaseMessageId) this.#acceptedMessageIds.delete(messageId);
if (this.#queues.get(key) === current) this.#queues.delete(key);
});
this.#queues.set(key, current);
return current;
}
async #handleMessageFailure(event, messageId, processingReaction, error) {
if (this.#signal?.aborted) {
await this.#removeProcessingReaction(messageId, processingReaction);
return;
}
this.#logger.error?.('[dsh-feishu] message handling failed:', error?.message ?? String(error));
this.#status.lastError = error?.message ?? String(error);
await this.#finishReaction(messageId, processingReaction, 'ERROR');
await this.#send(
event.message.chat_id,
'处理失败,请稍后重试。如果问题持续,请在 DeepSeek Harness 的飞书插件页面检查连接状态。',
).catch(() => undefined);
}
async waitForIdle() {
await Promise.allSettled([...this.#queues.values()]);
await Promise.allSettled([
...this.#queues.values(),
...[...this.#pendingInteractions.values()].flatMap((pending) => (
pending.queue ? [pending.queue] : []
)),
...this.#interactionTasks,
]);
}
async #handle(event, key) {
async #handle(event, key, { alreadyRecorded = false } = {}) {
this.#signal?.throwIfAborted();
const messageId = event.message.message_id;
await this.#state.markSeen(messageId);
this.#status.lastMessageAt = new Date().toISOString();
this.#status.messagesReceived += 1;
if (!alreadyRecorded) {
if (this.#state.hasSeen(messageId)) return;
await this.#state.markSeen(messageId);
this.#status.lastMessageAt = new Date().toISOString();
this.#status.messagesReceived += 1;
}
const text = extractText(event);
if (!text) return;
if (!text) {
await this.#send(event.message.chat_id, '目前仅支持文字消息。');
return;
}
if (text === '/help') {
await this.#send(event.message.chat_id, HELP_TEXT);
@ -108,7 +270,7 @@ export class FeishuHarnessBridge {
return;
}
if (text === '/status') {
await this.#harness.ensureRunning();
await this.#harness.ensureRunning({ signal: this.#signal });
await this.#send(event.message.chat_id, '飞书机器人与 DeepSeek Harness 连接正常。');
return;
}
@ -120,11 +282,29 @@ export class FeishuHarnessBridge {
return;
}
console.info(`[bridge] processing ${event.message.chat_type} message ${messageId}`);
await this.#answerWithStream(event, key, text);
this.#status.messagesReplied += 1;
this.#status.lastReplyAt = new Date().toISOString();
this.#status.lastError = null;
this.#logger.info?.(`[dsh-feishu] processing ${event.message.chat_type} message ${messageId}`);
try {
await this.#answerWithStream(event, key, text);
this.#status.messagesReplied += 1;
this.#status.lastReplyAt = new Date().toISOString();
this.#status.lastError = null;
} finally {
await this.#cancelPendingInteraction(key);
}
}
#interactionAskOptions(event, key) {
return {
timeoutMs: this.#replyTimeoutMs,
signal: this.#signal,
onInteraction: (interaction) => this.#handleInteraction(interaction, {
key,
actor: senderOpenId(event),
chatId: event.message.chat_id,
requiresMention: event.message.chat_type !== 'p2p',
}),
onInteractionResolved: (resolution) => this.#handleInteractionResolved(resolution),
};
}
async #answerWithStream(event, key, text) {
@ -136,7 +316,9 @@ export class FeishuHarnessBridge {
state: this.#state,
key,
text,
askOptions: { timeoutMs: this.#replyTimeoutMs },
createOptions: { signal: this.#signal },
existsOptions: { signal: this.#signal },
askOptions: this.#interactionAskOptions(event, key),
});
for (const chunk of splitText(answer)) await this.#send(chatId, chunk);
this.#status.streamFallbacks = (this.#status.streamFallbacks ?? 0) + 1;
@ -149,18 +331,21 @@ export class FeishuHarnessBridge {
await this.#channel.stream(chatId, {
markdown: async (controller) => {
promptStarted = true;
const askOptions = {
...this.#interactionAskOptions(event, key),
onUpdate: async (update) => {
await controller.setContent(this.#progressText(update));
this.#status.streamUpdates = (this.#status.streamUpdates ?? 0) + 1;
},
};
({ answer: completedAnswer } = await askInWorkspaceSession({
harness: this.#harness,
state: this.#state,
key,
text,
askOptions: {
timeoutMs: this.#replyTimeoutMs,
onUpdate: async (update) => {
await controller.setContent(this.#progressText(update));
this.#status.streamUpdates = (this.#status.streamUpdates ?? 0) + 1;
},
},
createOptions: { signal: this.#signal },
existsOptions: { signal: this.#signal },
askOptions,
}));
await controller.setContent(completedAnswer);
},
@ -169,26 +354,308 @@ export class FeishuHarnessBridge {
} catch (error) {
this.#status.streamErrors = (this.#status.streamErrors ?? 0) + 1;
if (completedAnswer) {
console.warn('[bridge] native Feishu stream failed after generation; sending final text:', error.message);
this.#logger.warn?.(
'[dsh-feishu] native stream failed after generation; sending final text:',
error.message,
);
for (const chunk of splitText(completedAnswer)) await this.#send(chatId, chunk);
this.#status.streamFallbacks = (this.#status.streamFallbacks ?? 0) + 1;
return;
}
if (promptStarted) throw error;
console.warn('[bridge] native Feishu stream unavailable; using text fallback:', error.message);
this.#logger.warn?.('[dsh-feishu] native stream unavailable; using text fallback:', error.message);
const { answer } = await askInWorkspaceSession({
harness: this.#harness,
state: this.#state,
key,
text,
askOptions: { timeoutMs: this.#replyTimeoutMs },
createOptions: { signal: this.#signal },
existsOptions: { signal: this.#signal },
askOptions: this.#interactionAskOptions(event, key),
});
for (const chunk of splitText(answer)) await this.#send(chatId, chunk);
this.#status.streamFallbacks = (this.#status.streamFallbacks ?? 0) + 1;
}
}
async #processInteractionReply(event, messageId, key, expected, processingReaction) {
this.#signal?.throwIfAborted();
const current = this.#pendingInteractions.get(key);
const claimed = expected.claimedReplyMessageId === messageId;
if (!current || current !== expected || current.submitting) {
if (this.#isResolvedQuestionReply(event, key)) {
return this.#discardResolvedInteractionReply(event, messageId);
}
if (claimed && (!current || current !== expected)) {
return this.#discardResolvedInteractionReply(event, messageId);
}
return this.#enqueueMessage(event, messageId, key, processingReaction, {
releaseMessageId: false,
finalize: false,
});
}
if (this.#state.hasSeen(messageId)) return;
await this.#state.markSeen(messageId);
this.#status.lastMessageAt = new Date().toISOString();
this.#status.messagesReceived += 1;
const text = extractText(event);
if (!text) {
await this.#send(event.message.chat_id, '请用文字回答当前问题。');
return;
}
const pending = this.#pendingInteractions.get(key);
if (!pending || pending !== expected || pending.submitting) {
if (this.#isResolvedQuestionReply(event, key)) {
await this.#send(event.message.chat_id, INTERACTION_RESOLVED_TEXT).catch(() => undefined);
return;
}
if (claimed && (!pending || pending !== expected)) {
await this.#send(event.message.chat_id, INTERACTION_RESOLVED_TEXT);
return;
}
return this.#enqueueMessage(event, messageId, key, processingReaction, {
releaseMessageId: false,
alreadyRecorded: true,
finalize: false,
});
}
pending.chatId = event.message.chat_id;
if (pending.needsPresentation) {
try {
await this.#presentInteraction(pending);
} catch {
this.#status.lastError = '飞书交互问题发送失败。';
this.#logger.error?.('[dsh-feishu] failed to retry an interaction question');
pending.interaction.reconnect?.();
}
return;
}
const question = pending.questions[pending.index];
if (!question) return;
pending.answers.push(harnessAnswerForQuestion(question, text));
pending.index += 1;
if (pending.index < pending.questions.length) {
if (pending.claimedReplyMessageId === messageId) {
pending.claimedReplyMessageId = null;
}
pending.needsPresentation = true;
try {
await this.#presentInteraction(pending);
} catch {
this.#status.lastError = '飞书交互问题发送失败。';
this.#logger.error?.('[dsh-feishu] failed to send the next interaction question');
pending.interaction.reconnect?.();
}
return;
}
pending.submitting = true;
try {
await pending.interaction.respond({
ok: true,
value: {
sessionId: pending.sessionId,
answer: { answers: pending.answers },
},
});
this.#rememberResolvedInteraction(key, pending);
this.#clearPendingInteraction(key, pending.interactionId);
this.#status.lastError = null;
} catch (error) {
if (this.#signal?.aborted) return;
if (this.#pendingInteractions.get(key) !== pending) return;
if (error?.code === 'interaction-not-pending') {
this.#rememberResolvedInteraction(key, pending);
this.#clearPendingInteraction(key, pending.interactionId);
await this.#send(event.message.chat_id, INTERACTION_RESOLVED_TEXT).catch(() => undefined);
return;
}
pending.submitting = false;
pending.answers.pop();
pending.index -= 1;
this.#status.lastError = '回答提交失败。';
this.#logger.error?.('[dsh-feishu] failed to answer a Harness interaction');
await this.#send(event.message.chat_id, '回答提交失败,请重新发送当前问题的答案。')
.catch(() => undefined);
}
}
async #handleInteraction(interaction, {
key,
actor,
chatId,
requiresMention,
}) {
// Approval is deliberately exposed by the transport but remains
// unanswered until #5 adds an authenticated policy and renderer.
if (interaction?.kind !== 'question') return;
const questions = interaction?.payload?.questions;
const interactionId = typeof interaction?.interactionId === 'string'
? interaction.interactionId
: interaction?.rpcId;
if (typeof interaction.rpcId !== 'string'
|| typeof interactionId !== 'string'
|| typeof interaction.sessionId !== 'string'
|| !Array.isArray(questions)
|| questions.length === 0
|| questions.some((question) => !validHarnessQuestion(question))) {
this.#logger.warn?.('[dsh-feishu] ignored an invalid Harness question interaction');
return;
}
if (interaction.recovered === true) {
await interaction.respond({
ok: false,
error: {
code: 'cancelled',
message: 'Feishu safely cancelled an interaction left by an earlier client.',
details: {},
},
});
await this.#send(
chatId,
'检测到这个 Session 中遗留的待回答问题,已安全取消并继续处理你刚才的消息。',
).catch(() => undefined);
return;
}
const existing = this.#pendingInteractions.get(key);
if (existing?.interactionId === interactionId) {
existing.interaction = interaction;
if (existing.needsPresentation) await this.#presentInteraction(existing);
return;
}
if (this.#interactionKeys.has(interactionId)) return;
if (existing) {
await interaction.respond({
ok: false,
error: {
code: 'cancelled',
message: 'Feishu is already handling another user interaction.',
details: {},
},
});
return;
}
const pending = {
kind: 'question',
interactionId,
sessionId: interaction.sessionId,
interaction,
key,
actor,
requiresMention,
questions,
answers: [],
index: 0,
chatId,
queue: null,
claimedReplyMessageId: null,
submitting: false,
needsPresentation: true,
questionMessageIds: new Set(),
inactive: false,
};
this.#pendingInteractions.set(key, pending);
this.#interactionKeys.set(pending.interactionId, key);
await this.#presentInteraction(pending);
}
#handleInteractionResolved(resolution) {
const interactionId = resolution?.interactionId;
if (resolution?.kind !== 'question' || typeof interactionId !== 'string') return;
const key = this.#interactionKeys.get(interactionId);
if (!key) return;
const pending = this.#pendingInteractions.get(key);
if (pending) this.#rememberResolvedInteraction(key, pending);
this.#clearPendingInteraction(key, interactionId);
}
async #presentInteraction(pending) {
const question = pending.questions[pending.index];
if (!question) return;
const messageId = await this.#send(
pending.chatId,
harnessQuestionText(
question,
pending.index,
pending.questions.length,
{ requiresMention: pending.requiresMention },
),
);
if (messageId) {
pending.questionMessageIds.add(messageId);
if (pending.inactive) this.#rememberResolvedInteraction(pending.key, pending);
}
pending.needsPresentation = false;
}
#rememberResolvedInteraction(key, pending) {
const expiresAt = Date.now() + RESOLVED_REPLY_TTL_MS;
for (const messageId of pending.questionMessageIds ?? []) {
this.#resolvedQuestionReplies.set(messageId, { key, expiresAt });
}
}
#isResolvedQuestionReply(event, key) {
const now = Date.now();
for (const [messageId, resolution] of this.#resolvedQuestionReplies) {
if (resolution.expiresAt <= now) this.#resolvedQuestionReplies.delete(messageId);
}
for (const reference of [event?.message?.parent_id, event?.message?.root_id]) {
const resolution = this.#resolvedQuestionReplies.get(reference);
if (resolution?.key === key && resolution.expiresAt > now) return true;
}
return false;
}
async #discardResolvedInteractionReply(event, messageId) {
if (this.#state.hasSeen(messageId)) return;
await this.#state.markSeen(messageId);
this.#status.lastMessageAt = new Date().toISOString();
this.#status.messagesReceived += 1;
await this.#send(event.message.chat_id, INTERACTION_RESOLVED_TEXT).catch(() => undefined);
}
#takePendingInteraction(key, interactionId) {
const pending = this.#pendingInteractions.get(key);
if (!pending
|| (interactionId !== undefined && pending.interactionId !== interactionId)) return null;
this.#pendingInteractions.delete(key);
this.#interactionKeys.delete(pending.interactionId);
pending.inactive = true;
return pending;
}
#clearPendingInteraction(key, interactionId) {
return this.#takePendingInteraction(key, interactionId) !== null;
}
async #cancelPendingInteraction(key) {
const pending = this.#takePendingInteraction(key);
if (!pending || pending.kind !== 'question') return;
this.#rememberResolvedInteraction(key, pending);
try {
await pending.interaction.respond({
ok: false,
error: {
code: 'cancelled',
message: 'The Feishu interaction ended before the user answered.',
details: {},
},
}, { signal: AbortSignal.timeout(5_000) });
} catch (error) {
if (error?.code !== 'interaction-not-pending') {
this.#logger.warn?.('[dsh-feishu] failed to cancel a pending Harness interaction');
}
}
}
#progressText(update) {
if (update.type === 'text' && update.text) return update.text;
if (update.type === 'tool') {
@ -206,12 +673,12 @@ export class FeishuHarnessBridge {
return reactionId;
} catch (error) {
this.#status.reactionErrors = (this.#status.reactionErrors ?? 0) + 1;
console.warn(`[bridge] unable to add ${emojiType} reaction:`, error.message);
this.#logger.warn?.(`[dsh-feishu] unable to add ${emojiType} reaction:`, error.message);
return null;
}
}
async #finishReaction(messageId, processingReaction, finalEmojiType) {
async #removeProcessingReaction(messageId, processingReaction) {
const reactionId = await processingReaction;
if (reactionId && this.#channel?.removeReaction) {
try {
@ -219,9 +686,13 @@ export class FeishuHarnessBridge {
this.#status.reactionsRemoved = (this.#status.reactionsRemoved ?? 0) + 1;
} catch (error) {
this.#status.reactionErrors = (this.#status.reactionErrors ?? 0) + 1;
console.warn('[bridge] unable to remove processing reaction:', error.message);
this.#logger.warn?.('[dsh-feishu] unable to remove processing reaction:', error.message);
}
}
}
async #finishReaction(messageId, processingReaction, finalEmojiType) {
await this.#removeProcessingReaction(messageId, processingReaction);
await this.#addReaction(messageId, finalEmojiType);
}
@ -237,5 +708,6 @@ export class FeishuHarnessBridge {
if (response?.code && response.code !== 0) {
throw new Error(`Feishu send failed: ${response.msg || response.code}`);
}
return nonEmptyString(response?.data?.message_id);
}
}

View file

@ -1,6 +1,26 @@
import { FeishuHarnessBridge } from './bridge.mjs';
import { VerifiedFeishuChannel } from './feishu-channel.mjs';
const DEFAULT_REQUEST_TIMEOUT_MS = 15_000;
function httpInstanceWithTimeout(httpInstance, timeoutMs) {
if (!httpInstance || typeof httpInstance.request !== 'function') return undefined;
const optionsWithTimeout = (options) => ({
...(options ?? {}),
timeout: options?.timeout ?? timeoutMs,
});
return {
request: (options) => httpInstance.request(optionsWithTimeout(options)),
get: (url, options) => httpInstance.get(url, optionsWithTimeout(options)),
delete: (url, options) => httpInstance.delete(url, optionsWithTimeout(options)),
head: (url, options) => httpInstance.head(url, optionsWithTimeout(options)),
options: (url, options) => httpInstance.options(url, optionsWithTimeout(options)),
post: (url, data, options) => httpInstance.post(url, data, optionsWithTimeout(options)),
put: (url, data, options) => httpInstance.put(url, data, optionsWithTimeout(options)),
patch: (url, data, options) => httpInstance.patch(url, data, optionsWithTimeout(options)),
};
}
export function createBridgeStatus({ allowedSenderCount = 1 } = {}) {
return {
startedAt: null,
@ -43,11 +63,13 @@ export class FeishuRuntime {
#state;
#replyTimeoutMs;
#connectTimeoutMs;
#requestTimeoutMs;
#logger;
#client = null;
#bridge = null;
#wsClient = null;
#starting = null;
#abortController = null;
#status;
constructor({
@ -61,6 +83,7 @@ export class FeishuRuntime {
state,
replyTimeoutMs = 600000,
connectTimeoutMs = 15000,
requestTimeoutMs = DEFAULT_REQUEST_TIMEOUT_MS,
logger = console,
}) {
if (!lark) throw new Error('FeishuRuntime requires the Feishu SDK');
@ -70,6 +93,9 @@ export class FeishuRuntime {
if (normalizedOwners.length === 0) throw new Error('FeishuRuntime requires at least one owner open_id');
if (!harness) throw new Error('FeishuRuntime requires a Harness client');
if (!state) throw new Error('FeishuRuntime requires a state store');
if (!Number.isFinite(requestTimeoutMs) || requestTimeoutMs <= 0) {
throw new TypeError('FeishuRuntime requestTimeoutMs must be a positive number');
}
this.#lark = lark;
this.#appId = appId;
@ -80,6 +106,7 @@ export class FeishuRuntime {
this.#state = state;
this.#replyTimeoutMs = replyTimeoutMs;
this.#connectTimeoutMs = connectTimeoutMs;
this.#requestTimeoutMs = requestTimeoutMs;
this.#logger = logger;
this.#status = createBridgeStatus({ allowedSenderCount: normalizedOwners.length });
}
@ -99,12 +126,15 @@ export class FeishuRuntime {
}
async #start() {
const abortController = new AbortController();
this.#abortController = abortController;
const { signal } = abortController;
this.#status.startedAt = new Date().toISOString();
this.#status.feishuLongConnectionState = 'connecting';
this.#status.lastError = null;
try {
await this.#harness.ensureRunning();
await this.#harness.ensureRunning({ signal });
this.#status.harnessReachable = true;
const sdkDomain = this.#domain === 'lark'
@ -115,6 +145,11 @@ export class FeishuRuntime {
appSecret: this.#appSecret,
domain: sdkDomain,
};
const httpInstance = httpInstanceWithTimeout(
this.#lark.defaultHttpInstance,
this.#requestTimeoutMs,
);
if (httpInstance) larkConfig.httpInstance = httpInstance;
this.#client = new this.#lark.Client(larkConfig);
const channel = new VerifiedFeishuChannel({
client: this.#client,
@ -128,6 +163,8 @@ export class FeishuRuntime {
status: this.#status,
allowedSenderOpenIds: new Set(this.#ownerOpenIds),
replyTimeoutMs: this.#replyTimeoutMs,
signal,
logger: this.#logger,
});
const dispatcher = new this.#lark.EventDispatcher({}).register({
@ -205,6 +242,9 @@ export class FeishuRuntime {
async stop({ preserveError = false } = {}) {
const error = preserveError ? this.#status.lastError : null;
const abortController = this.#abortController;
this.#abortController = null;
abortController?.abort(new DOMException('Feishu runtime stopped', 'AbortError'));
this.#status.ready = false;
if (this.#wsClient) {
this.#wsClient.close({ force: true });

View file

@ -1,338 +1,19 @@
import { spawn } from 'node:child_process';
import { randomUUID } from 'node:crypto';
import { isAbsolute } from 'node:path';
import {
HarnessClient as SharedHarnessClient,
} from '../shared/harness-client.mjs';
import { adoptRegisteredWorkspaceSession } from '../shared/harness-session-binding.mjs';
export {
HarnessInteractionError,
HarnessReplyTracker,
HarnessRpcError,
} from '../shared/harness-client.mjs';
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
function workspacePaths(value) {
if (!Array.isArray(value?.items)) return [];
return value.items.flatMap((item) => (
typeof item?.path === 'string' && isAbsolute(item.path) ? [item.path] : []
));
}
function workspaceFromList(workspacePath, workspaceList) {
if (!Array.isArray(workspaceList?.items)
|| !Array.isArray(workspaceList?.archivedSessionIds)) {
throw new Error('Harness returned an invalid response for workspace.list');
}
const workspace = workspaceList.items.find((item) => item?.path === workspacePath);
if (!workspace) return null;
if (!Array.isArray(workspace.sessionIds)
|| workspace.sessionIds.some((sessionId) => typeof sessionId !== 'string')) {
throw new Error('Harness returned invalid session IDs for workspace.list');
}
return workspace;
}
function workspaceSessions(workspace, archivedSessionIds, sessionList) {
if (!Array.isArray(sessionList?.items)) {
throw new Error('Harness returned an invalid response for session.list');
}
const archived = new Set(archivedSessionIds);
const summaries = new Map(sessionList.items.flatMap((item) => (
typeof item?.sessionId === 'string' ? [[item.sessionId, item]] : []
)));
return {
workspace: workspace.path,
sessions: workspace.sessionIds.map((sessionId) => {
const summary = summaries.get(sessionId);
const title = summary?.projections?.values?.title;
return {
sessionId,
title: typeof title === 'string' ? title : null,
archived: archived.has(sessionId),
blank: summary?.blank === true,
origin: summary?.origin === 'subagent' ? 'subagent' : null,
summaryAvailable: summary !== undefined,
};
}),
};
}
function messageText(event) {
return (event?.data?.message?.content ?? [])
.filter((part) => part.type === 'text' && typeof part.text === 'string')
.map((part) => part.text)
.join('\n')
.trim();
}
export class HarnessReplyTracker {
#promptRpcId;
#lastSeq;
#openTurn = null;
#targetTurn = null;
#stepText = new Map();
#latestText = '';
#finished = false;
#reason = null;
constructor({ promptRpcId, afterSeq = -1 }) {
this.#promptRpcId = promptRpcId;
this.#lastSeq = afterSeq;
}
get finished() {
return this.#finished;
}
get answer() {
return this.#latestText.trim();
}
get reason() {
return this.#reason;
}
consume(entries) {
let update = null;
const ordered = [...entries]
.map((entry) => entry?.event ?? entry)
.filter(Boolean)
.sort((left, right) => (left.seq ?? -1) - (right.seq ?? -1));
for (const event of ordered) {
const seq = event.seq ?? -1;
if (seq <= this.#lastSeq) continue;
this.#lastSeq = seq;
if (event.type === 'turn/start') {
this.#openTurn = event.data?.turn ?? null;
}
if (event.type === 'user/message'
&& event.data?.source?.rpcId === this.#promptRpcId) {
this.#targetTurn = this.#openTurn;
continue;
}
if (this.#targetTurn === null) continue;
if (event.type === 'turn/end') {
if (event.data?.turn !== this.#targetTurn) continue;
this.#finished = true;
this.#reason = event.data?.reason ?? null;
this.#openTurn = null;
continue;
}
if (event.data?.turn !== this.#targetTurn) continue;
if (event.type === 'assistant/chunk' && event.data?.chunk?.type === 'text-delta') {
const step = event.data?.step ?? 0;
const index = event.data.chunk.index ?? 0;
const key = `${step}:${index}`;
this.#stepText.set(key, (this.#stepText.get(key) ?? '') + event.data.chunk.text);
const stepPrefix = `${step}:`;
const text = [...this.#stepText.entries()]
.filter(([partKey]) => partKey.startsWith(stepPrefix))
.sort(([left], [right]) => Number(left.split(':')[1]) - Number(right.split(':')[1]))
.map(([, part]) => part)
.join('\n')
.trim();
if (text && text !== this.#latestText) {
this.#latestText = text;
update = { type: 'text', text };
}
continue;
}
if (event.type === 'assistant/message') {
const text = messageText(event);
if (text && text !== this.#latestText) {
this.#latestText = text;
update = { type: 'text', text };
}
continue;
}
if (event.type === 'tool/call') {
update = { type: 'tool', name: event.data?.name ?? '工具' };
} else if (event.type === 'tool/result') {
update = { type: 'status', text: '正在整理结果…' };
}
}
return update;
}
}
export class HarnessRpcError extends Error {
constructor(method, error) {
super(`${method}: ${error?.message ?? 'unknown Harness RPC error'}`);
this.name = 'HarnessRpcError';
this.method = method;
this.code = error?.code ?? 'internal';
this.details = error?.details ?? {};
}
}
export class HarnessClient {
#baseUrl;
#workspace;
#agentPreset;
#autostart;
#dshBin;
#managedProcess = null;
constructor({ baseUrl, workspace, agentPreset, autostart, dshBin }) {
this.#baseUrl = new URL(baseUrl);
this.#workspace = workspace;
this.#agentPreset = agentPreset;
this.#autostart = autostart;
this.#dshBin = dshBin;
}
async rpc(method, payload = {}, timeoutMs = 30000, options = {}) {
const rpcId = options.rpcId ?? `feishu-${randomUUID()}`;
const response = await fetch(new URL(`/api/${method}`, this.#baseUrl), {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify({ type: 'client-request', rpcId, method, payload }),
signal: AbortSignal.timeout(timeoutMs),
export class HarnessClient extends SharedHarnessClient {
constructor(options) {
super({
...options,
rpcIdPrefix: 'feishu',
logPrefix: 'dsh-feishu',
});
if (!response.ok) throw new Error(`Harness transport ${method} failed: HTTP ${response.status}`);
const body = await response.json();
if (body?.type !== 'server-response' || body?.rpcId !== rpcId) {
throw new Error(`Harness returned an invalid response for ${method}`);
}
if (!body.result?.ok) throw new HarnessRpcError(method, body.result?.error);
return body.result.value;
}
async health() {
await this.rpc('host.describe', {}, 5000);
return true;
}
async ensureRunning() {
try {
return await this.health();
} catch (firstError) {
if (!this.#autostart) throw firstError;
}
if (!this.#managedProcess || this.#managedProcess.exitCode !== null) {
const port = this.#baseUrl.port || (this.#baseUrl.protocol === 'https:' ? '443' : '80');
this.#managedProcess = spawn(this.#dshBin, [
'web',
'--host',
this.#baseUrl.hostname,
'--port',
port,
], {
cwd: this.#workspace,
env: process.env,
stdio: ['ignore', 'inherit', 'inherit'],
});
this.#managedProcess.on('error', (error) => {
console.error('[bridge] failed to start Harness:', error.message);
});
}
const deadline = Date.now() + 60000;
let lastError;
while (Date.now() < deadline) {
await sleep(1000);
try {
return await this.health();
} catch (error) {
lastError = error;
}
}
throw new Error(`Harness did not become ready: ${lastError?.message ?? 'timeout'}`);
}
async listWorkspaces(options = {}) {
await this.ensureRunning();
return workspacePaths(await this.rpc('workspace.list', {}, 30000, options));
}
async listWorkspaceSessions(workspacePath, options = {}) {
await this.ensureRunning();
const workspaceList = await this.rpc('workspace.list', {}, 30000, options);
const workspace = workspaceFromList(workspacePath, workspaceList);
if (!workspace) return { workspace: workspacePath, sessions: [] };
const sessionList = await this.rpc('session.list', {}, 30000, options);
return workspaceSessions(workspace, workspaceList.archivedSessionIds, sessionList);
}
async adoptWorkspaceSession(value, options = {}) {
return adoptRegisteredWorkspaceSession(this, value, options, 30000);
}
async workspaceId(options = {}) {
const workspace = options.workspace ?? this.#workspace;
const { items } = await this.rpc('workspace.list', {});
const existing = items.find((item) => item.path === workspace);
if (existing) return existing.workspaceId;
const created = await this.rpc('workspace.create', { path: workspace });
return created.workspace.workspaceId;
}
async createSession(options = {}) {
await this.ensureRunning();
const workspaceId = await this.workspaceId(options);
const created = await this.rpc('session.create', {
workspaceId,
agentPreset: this.#agentPreset,
});
return created.sessionId;
}
async sessionExists(sessionId) {
try {
await this.rpc('session.history', { sessionId, maxMessages: 1 });
return true;
} catch (error) {
if (error instanceof HarnessRpcError && error.code === 'session-not-found') return false;
throw error;
}
}
async ask(sessionId, text, options = {}) {
if (typeof options === 'number') options = { timeoutMs: options };
const timeoutMs = options.timeoutMs ?? 600000;
const onUpdate = typeof options.onUpdate === 'function' ? options.onUpdate : null;
await this.ensureRunning();
const before = await this.rpc('session.history', { sessionId, maxMessages: 1 });
const baselineSeq = Math.max(-1, ...(before.events ?? []).map(({ event }) => event.seq ?? -1));
const promptRpcId = `feishu-${randomUUID()}`;
const tracker = new HarnessReplyTracker({ promptRpcId, afterSeq: baselineSeq });
await this.rpc('session.prompt', {
sessionId,
mode: 'queue',
content: [{ type: 'text', text }],
clientTimeZone: Intl.DateTimeFormat().resolvedOptions().timeZone,
}, 30000, { rpcId: promptRpcId });
const deadline = Date.now() + timeoutMs;
while (Date.now() < deadline) {
await sleep(300);
const history = await this.rpc('session.history', { sessionId, maxMessages: 50 });
const update = tracker.consume(history.events ?? []);
if (update && onUpdate) {
try {
await onUpdate(update);
} catch (error) {
console.warn('[bridge] ignored a progress update failure:', error.message);
}
}
if (!tracker.finished) continue;
if (tracker.answer) return tracker.answer;
throw new Error(`Harness turn ended without a text reply${tracker.reason ? ` (${JSON.stringify(tracker.reason)})` : ''}`);
}
throw new Error(`Harness reply timed out after ${Math.round(timeoutMs / 1000)} seconds`);
}
stopManagedProcess() {
if (this.#managedProcess?.exitCode === null) this.#managedProcess.kill('SIGTERM');
}
}

View file

@ -0,0 +1,825 @@
import { spawn } from 'node:child_process';
import { randomUUID } from 'node:crypto';
import { isAbsolute } from 'node:path';
import { adoptRegisteredWorkspaceSession } from './harness-session-binding.mjs';
// Every channel plugin runs in the same Host process. Sharing ownership by
// Harness origin prevents two channel-specific clients bound to one Session
// from claiming or cancelling each other's interactions.
const interactionRegistries = new Map();
function interactionRegistry(origin) {
let registry = interactionRegistries.get(origin);
if (!registry) {
registry = { ownerships: new Map(), claims: new Map(), nextOrder: 0 };
interactionRegistries.set(origin, registry);
}
return registry;
}
function workspacePaths(value) {
if (!Array.isArray(value?.items)) return [];
return value.items.flatMap((item) => (
typeof item?.path === 'string' && isAbsolute(item.path) ? [item.path] : []
));
}
function workspaceFromList(workspacePath, workspaceList) {
if (!Array.isArray(workspaceList?.items)
|| !Array.isArray(workspaceList?.archivedSessionIds)) {
throw new Error('Harness returned an invalid response for workspace.list');
}
const workspace = workspaceList.items.find((item) => item?.path === workspacePath);
if (!workspace) return null;
if (!Array.isArray(workspace.sessionIds)
|| workspace.sessionIds.some((sessionId) => typeof sessionId !== 'string')) {
throw new Error('Harness returned invalid session IDs for workspace.list');
}
return workspace;
}
function workspaceSessions(workspace, archivedSessionIds, sessionList) {
if (!Array.isArray(sessionList?.items)) {
throw new Error('Harness returned an invalid response for session.list');
}
const archived = new Set(archivedSessionIds);
const summaries = new Map(sessionList.items.flatMap((item) => (
typeof item?.sessionId === 'string' ? [[item.sessionId, item]] : []
)));
return {
workspace: workspace.path,
sessions: workspace.sessionIds.map((sessionId) => {
const summary = summaries.get(sessionId);
const title = summary?.projections?.values?.title;
return {
sessionId,
title: typeof title === 'string' ? title : null,
archived: archived.has(sessionId),
blank: summary?.blank === true,
origin: summary?.origin === 'subagent' ? 'subagent' : null,
summaryAvailable: summary !== undefined,
};
}),
};
}
function sleep(ms, signal) {
return new Promise((resolve, reject) => {
if (signal?.aborted) {
reject(signal.reason ?? new DOMException('Aborted', 'AbortError'));
return;
}
const timer = setTimeout(() => {
signal?.removeEventListener('abort', onAbort);
resolve();
}, ms);
const onAbort = () => {
clearTimeout(timer);
reject(signal.reason ?? new DOMException('Aborted', 'AbortError'));
};
signal?.addEventListener('abort', onAbort, { once: true });
});
}
function assistantMessageText(event) {
return (event?.data?.message?.content ?? [])
.filter((part) => part.type === 'text' && typeof part.text === 'string')
.map((part) => part.text)
.join('\n')
.trim();
}
function consumeInteractionOwnership(ownership, entries) {
const ordered = [...entries]
.map((entry) => entry?.event ?? entry)
.filter(Boolean)
.sort((left, right) => (left.seq ?? -1) - (right.seq ?? -1));
for (const event of ordered) {
const seq = event.seq ?? -1;
if (seq <= ownership.lastSeq) continue;
ownership.lastSeq = seq;
if (event.type === 'turn/start') {
const turn = event.data?.turn ?? null;
if (ownership.active && turn !== ownership.turn) ownership.active = false;
if (ownership.turn !== null && turn !== ownership.turn) ownership.completed = true;
ownership.openTurn = turn;
continue;
}
if (event.type === 'user/message' && event.data?.source?.rpcId === ownership.promptRpcId) {
ownership.active = true;
ownership.started = true;
ownership.completed = false;
ownership.turn = event.data?.turn ?? ownership.openTurn;
continue;
}
if (event.type === 'turn/end' && event.data?.turn === ownership.turn) {
ownership.active = false;
ownership.completed = true;
}
}
}
export class HarnessReplyTracker {
#promptRpcId;
#lastSeq;
#openTurn = null;
#targetTurn = null;
#stepText = new Map();
#latestText = '';
#finished = false;
#reason = null;
constructor({ promptRpcId, afterSeq = -1 }) {
this.#promptRpcId = promptRpcId;
this.#lastSeq = afterSeq;
}
get finished() {
return this.#finished;
}
get answer() {
return this.#latestText.trim();
}
get reason() {
return this.#reason;
}
get tracking() {
return this.#targetTurn !== null && !this.#finished;
}
get turn() {
return this.#targetTurn;
}
consume(entries) {
let update = null;
const ordered = [...entries]
.map((entry) => entry?.event ?? entry)
.filter(Boolean)
.sort((left, right) => (left.seq ?? -1) - (right.seq ?? -1));
for (const event of ordered) {
const seq = event.seq ?? -1;
if (seq <= this.#lastSeq) continue;
this.#lastSeq = seq;
if (event.type === 'turn/start') this.#openTurn = event.data?.turn ?? null;
if (event.type === 'user/message' && event.data?.source?.rpcId === this.#promptRpcId) {
this.#targetTurn = this.#openTurn;
continue;
}
if (this.#targetTurn === null) continue;
if (event.type === 'turn/end') {
if (event.data?.turn !== this.#targetTurn) continue;
this.#finished = true;
this.#reason = event.data?.reason ?? null;
this.#openTurn = null;
continue;
}
if (event.data?.turn !== this.#targetTurn) continue;
if (event.type === 'assistant/chunk' && event.data?.chunk?.type === 'text-delta') {
const step = event.data?.step ?? 0;
const index = event.data.chunk.index ?? 0;
const key = `${step}:${index}`;
this.#stepText.set(key, (this.#stepText.get(key) ?? '') + event.data.chunk.text);
const prefix = `${step}:`;
const text = [...this.#stepText.entries()]
.filter(([partKey]) => partKey.startsWith(prefix))
.sort(([left], [right]) => Number(left.split(':')[1]) - Number(right.split(':')[1]))
.map(([, part]) => part)
.join('\n')
.trim();
if (text && text !== this.#latestText) {
this.#latestText = text;
update = { type: 'text', text };
}
continue;
}
if (event.type === 'assistant/message') {
const text = assistantMessageText(event);
if (text && text !== this.#latestText) {
this.#latestText = text;
update = { type: 'text', text };
}
continue;
}
if (event.type === 'tool/call') {
update = { type: 'tool', name: event.data?.name ?? '工具' };
} else if (event.type === 'tool/result') {
update = { type: 'status', text: '正在整理结果…' };
}
}
return update;
}
}
export class HarnessRpcError extends Error {
constructor(method, error) {
super(`${method}: ${error?.message ?? 'unknown Harness RPC error'}`);
this.name = 'HarnessRpcError';
this.method = method;
this.code = error?.code ?? 'internal';
this.details = error?.details ?? {};
}
}
export class HarnessInteractionError extends Error {
constructor(code, message) {
super(message);
this.name = 'HarnessInteractionError';
this.code = code;
}
}
export class HarnessClient {
#baseUrl;
#workspace;
#agentPreset;
#autostart;
#dshBin;
#fetch;
#createWebSocket;
#interactionReconnectDelayMs;
#rpcIdPrefix;
#logPrefix;
#managedProcess = null;
#interactionRegistry;
#interactionOwnerships;
#interactionClaims;
constructor({
baseUrl,
workspace,
agentPreset = 'standard',
autostart = false,
dshBin = 'dsh',
fetchImpl = fetch,
createWebSocket = (url) => new WebSocket(url),
interactionReconnectDelayMs = 500,
rpcIdPrefix = 'im',
logPrefix = 'dsh-im',
}) {
if (typeof createWebSocket !== 'function') {
throw new TypeError('createWebSocket must be a function');
}
if (!Number.isFinite(interactionReconnectDelayMs) || interactionReconnectDelayMs < 0) {
throw new TypeError('interactionReconnectDelayMs must be a non-negative number');
}
if (typeof rpcIdPrefix !== 'string' || !rpcIdPrefix.trim()) {
throw new TypeError('rpcIdPrefix must be a non-empty string');
}
if (typeof logPrefix !== 'string' || !logPrefix.trim()) {
throw new TypeError('logPrefix must be a non-empty string');
}
this.#baseUrl = new URL(baseUrl);
this.#workspace = workspace;
this.#agentPreset = agentPreset;
this.#autostart = autostart;
this.#dshBin = dshBin;
this.#fetch = fetchImpl;
this.#createWebSocket = createWebSocket;
this.#interactionReconnectDelayMs = interactionReconnectDelayMs;
this.#rpcIdPrefix = rpcIdPrefix.trim();
this.#logPrefix = logPrefix.trim();
this.#interactionRegistry = interactionRegistry(this.#baseUrl.origin);
this.#interactionOwnerships = this.#interactionRegistry.ownerships;
this.#interactionClaims = this.#interactionRegistry.claims;
}
async rpc(method, payload = {}, timeoutMs = 30_000, options = {}) {
const rpcId = options.rpcId ?? `${this.#rpcIdPrefix}-${randomUUID()}`;
const timeoutSignal = AbortSignal.timeout(timeoutMs);
const signal = options.signal
? AbortSignal.any([options.signal, timeoutSignal])
: timeoutSignal;
const response = await this.#fetch(new URL(`/api/${method}`, this.#baseUrl), {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify({ type: 'client-request', rpcId, method, payload }),
signal,
});
if (!response.ok) throw new Error(`Harness transport ${method} failed: HTTP ${response.status}`);
const body = await response.json();
if (body?.type !== 'server-response' || body?.rpcId !== rpcId) {
throw new Error(`Harness returned an invalid response for ${method}`);
}
if (!body.result?.ok) throw new HarnessRpcError(method, body.result?.error);
return body.result.value;
}
async health(options = {}) {
await this.rpc('host.describe', {}, 5_000, options);
return true;
}
async ensureRunning(options = {}) {
try {
return await this.health(options);
} catch (firstError) {
if (!this.#autostart) throw firstError;
}
if (!this.#managedProcess || this.#managedProcess.exitCode !== null) {
const port = this.#baseUrl.port || (this.#baseUrl.protocol === 'https:' ? '443' : '80');
this.#managedProcess = spawn(this.#dshBin, [
'web', '--host', this.#baseUrl.hostname, '--port', port,
], {
cwd: this.#workspace,
env: process.env,
stdio: ['ignore', 'inherit', 'inherit'],
});
this.#managedProcess.on('error', (error) => {
console.error(`[${this.#logPrefix}] failed to start Harness:`, error.message);
});
}
const deadline = Date.now() + 60_000;
let lastError;
while (Date.now() < deadline) {
await sleep(1_000, options.signal);
try {
return await this.health(options);
} catch (error) {
lastError = error;
}
}
throw new Error(`Harness did not become ready: ${lastError?.message ?? 'timeout'}`);
}
async listWorkspaces(options = {}) {
await this.ensureRunning(options);
return workspacePaths(await this.rpc('workspace.list', {}, 30_000, options));
}
async listWorkspaceSessions(workspacePath, options = {}) {
await this.ensureRunning(options);
const workspaceList = await this.rpc('workspace.list', {}, 30_000, options);
const workspace = workspaceFromList(workspacePath, workspaceList);
if (!workspace) return { workspace: workspacePath, sessions: [] };
const sessionList = await this.rpc('session.list', {}, 30_000, options);
return workspaceSessions(workspace, workspaceList.archivedSessionIds, sessionList);
}
async adoptWorkspaceSession(value, options = {}) {
return adoptRegisteredWorkspaceSession(this, value, options);
}
async workspaceId(options = {}) {
const { workspace = this.#workspace, ...rpcOptions } = options;
const { items } = await this.rpc('workspace.list', {}, 30_000, rpcOptions);
const existing = items.find((item) => item.path === workspace);
if (existing) return existing.workspaceId;
const created = await this.rpc('workspace.create', { path: workspace }, 30_000, rpcOptions);
return created.workspace.workspaceId;
}
async createSession(options = {}) {
await this.ensureRunning(options);
const workspaceId = await this.workspaceId(options);
const created = await this.rpc('session.create', {
workspaceId,
agentPreset: this.#agentPreset,
}, 30_000, options);
return created.sessionId;
}
async sessionExists(sessionId, options = {}) {
try {
await this.rpc('session.history', { sessionId, maxMessages: 1 }, 30_000, options);
return true;
} catch (error) {
if (error instanceof HarnessRpcError && error.code === 'session-not-found') return false;
throw error;
}
}
async respondInteraction(rpcId, result, options = {}) {
if (typeof rpcId !== 'string' || !rpcId) throw new TypeError('rpcId is required');
if (!result || typeof result !== 'object' || typeof result.ok !== 'boolean') {
throw new TypeError('A Harness RPC result is required');
}
const timeoutSignal = AbortSignal.timeout(options.timeoutMs ?? 30_000);
const signal = options.signal
? AbortSignal.any([options.signal, timeoutSignal])
: timeoutSignal;
const response = await this.#fetch(new URL('/api/respond', this.#baseUrl), {
method: 'POST',
headers: { 'content-type': 'application/json' },
body: JSON.stringify({ type: 'client-response', rpcId, result }),
signal,
});
if (!response.ok) {
throw new Error(`Harness transport respond failed: HTTP ${response.status}`);
}
const receipt = await response.json();
if (receipt?.accepted === true) return receipt;
if (receipt?.accepted !== false
|| (receipt.reason !== 'bad-response' && receipt.reason !== 'not-pending')) {
throw new Error('Harness returned an invalid interaction response receipt');
}
const reason = receipt.reason;
throw new HarnessInteractionError(
`interaction-${reason}`,
`Harness interaction response was rejected (${reason})`,
);
}
async watchInteractions(sessionId, {
signal,
onInteraction,
onResolved,
onOpen,
ownership,
} = {}) {
if (typeof sessionId !== 'string' || !sessionId) throw new TypeError('sessionId is required');
if (!signal || typeof signal.addEventListener !== 'function') {
throw new TypeError('watchInteractions requires an AbortSignal');
}
if (onInteraction !== undefined && typeof onInteraction !== 'function') {
throw new TypeError('onInteraction must be a function');
}
if (onResolved !== undefined && typeof onResolved !== 'function') {
throw new TypeError('onResolved must be a function');
}
if (onOpen !== undefined && typeof onOpen !== 'function') {
throw new TypeError('onOpen must be a function');
}
while (!signal.aborted) {
try {
await this.#watchInteractionSocket(sessionId, {
signal,
onInteraction,
onResolved,
onOpen,
ownership,
});
} catch (error) {
if (signal.aborted) return;
console.warn(`[${this.#logPrefix}] Harness interaction stream disconnected:`, error.message);
}
if (signal.aborted) return;
try {
await sleep(this.#interactionReconnectDelayMs, signal);
} catch {
if (signal.aborted) return;
throw new Error('Harness interaction reconnect wait failed');
}
}
}
#registerInteractionOwnership(sessionId, ownership) {
const owners = this.#interactionOwnerships.get(sessionId) ?? new Set();
ownership.order = this.#interactionRegistry.nextOrder;
this.#interactionRegistry.nextOrder += 1;
owners.add(ownership);
this.#interactionOwnerships.set(sessionId, owners);
}
#unregisterInteractionOwnership(sessionId, ownership) {
const owners = this.#interactionOwnerships.get(sessionId);
owners?.delete(ownership);
if (owners?.size === 0) this.#interactionOwnerships.delete(sessionId);
for (const [key, claim] of this.#interactionClaims) {
if (claim.ownership === ownership) this.#interactionClaims.delete(key);
}
}
#consumeInteractionOwnerships(sessionId, entries) {
for (const ownership of this.#interactionOwnerships.get(sessionId) ?? []) {
consumeInteractionOwnership(ownership, entries);
}
}
async #refreshInteractionOwnerships(sessionId, signal) {
const history = await this.rpc(
'session.history',
{ sessionId, maxMessages: 50 },
30_000,
{ signal },
);
this.#consumeInteractionOwnerships(sessionId, history.events ?? []);
}
#interactionOwner(sessionId, claimKey, kind) {
const claim = this.#interactionClaims.get(claimKey);
if (claim && this.#interactionOwnerships.get(sessionId)?.has(claim.ownership)) return claim;
const owners = [...(this.#interactionOwnerships.get(sessionId) ?? [])];
const active = owners
.filter((ownership) => ownership.active)
.sort((left, right) => left.order - right.order);
if (active.length > 0) return { ownership: active[0], recovered: false };
// Approval recovery must remain fail-closed until #5 can prove the actor
// and conversation that originally requested the decision.
if (kind !== 'question') return null;
// A newly attached IM conversation may encounter a question left by
// an earlier runtime before its queued prompt starts. Let the oldest such
// ask adopt that replay so the Session can recover instead of deadlocking.
const ownership = owners
.filter((ownership) => !ownership.started && !ownership.completed)
.sort((left, right) => left.order - right.order)[0] ?? null;
return ownership ? { ownership, recovered: true } : null;
}
async ask(sessionId, text, options = {}) {
if (typeof options === 'number') options = { timeoutMs: options };
const timeoutMs = options.timeoutMs ?? 600_000;
const signal = options.signal;
const onUpdate = typeof options.onUpdate === 'function' ? options.onUpdate : null;
const onInteraction = typeof options.onInteraction === 'function'
? options.onInteraction
: undefined;
const onInteractionResolved = typeof options.onInteractionResolved === 'function'
? options.onInteractionResolved
: undefined;
await this.ensureRunning({ signal });
const before = await this.rpc(
'session.history',
{ sessionId, maxMessages: 1 },
30_000,
{ signal },
);
const baselineSeq = Math.max(-1, ...(before.events ?? []).map(({ event }) => event.seq ?? -1));
const promptRpcId = `${this.#rpcIdPrefix}-${randomUUID()}`;
const tracker = new HarnessReplyTracker({ promptRpcId, afterSeq: baselineSeq });
const interactionController = onInteraction || onInteractionResolved
? new AbortController()
: null;
const interactionSignal = interactionController
? (signal
? AbortSignal.any([signal, interactionController.signal])
: interactionController.signal)
: null;
// The mux is host-global. A prompt RPC becomes the owner only when its
// durable user/message starts a turn, so two chats bound to one Session
// cannot answer each other's questions or approvals.
const ownership = interactionController
? {
promptRpcId,
active: false,
started: false,
completed: false,
turn: null,
openTurn: null,
lastSeq: baselineSeq,
reconnect: null,
order: -1,
}
: null;
let interactionTask = null;
if (ownership) this.#registerInteractionOwnership(sessionId, ownership);
try {
if (interactionSignal) {
let markOpen;
const opened = new Promise((resolve) => { markOpen = resolve; });
interactionTask = this.watchInteractions(sessionId, {
signal: interactionSignal,
onInteraction,
onResolved: onInteractionResolved,
onOpen: markOpen,
ownership,
});
void interactionTask.catch(() => undefined);
await Promise.race([
opened,
sleep(30_000, interactionSignal).then(() => {
throw new Error('Harness interaction stream did not open within 30 seconds');
}),
]);
}
await this.rpc('session.prompt', {
sessionId,
mode: 'queue',
content: [{ type: 'text', text }],
clientTimeZone: Intl.DateTimeFormat().resolvedOptions().timeZone,
}, 30_000, { rpcId: promptRpcId, signal });
const deadline = Date.now() + timeoutMs;
while (Date.now() < deadline) {
await sleep(300, signal);
const history = await this.rpc(
'session.history',
{ sessionId, maxMessages: 50 },
30_000,
{ signal },
);
const wasActive = ownership?.active === true;
if (ownership) {
this.#consumeInteractionOwnerships(sessionId, history.events ?? []);
if (!wasActive && ownership.active) ownership.reconnect?.();
}
const update = tracker.consume(history.events ?? []);
if (update && onUpdate) {
try {
await onUpdate(update);
} catch (error) {
console.warn(`[${this.#logPrefix}] ignored a progress update failure:`, error.message);
}
}
if (!tracker.finished) continue;
if (tracker.answer) return tracker.answer;
throw new Error(
`Harness turn ended without a text reply${tracker.reason ? ` (${JSON.stringify(tracker.reason)})` : ''}`,
);
}
throw new Error(`Harness reply timed out after ${Math.round(timeoutMs / 1_000)} seconds`);
} finally {
interactionController?.abort(new DOMException('Harness turn finished', 'AbortError'));
if (interactionTask) await interactionTask.catch(() => undefined);
if (ownership) this.#unregisterInteractionOwnership(sessionId, ownership);
}
}
#watchInteractionSocket(sessionId, {
signal,
onInteraction,
onResolved,
onOpen,
ownership,
}) {
const url = new URL('/api/events.mux', this.#baseUrl);
url.protocol = url.protocol === 'https:' ? 'wss:' : 'ws:';
return new Promise((resolve, reject) => {
let socket;
try {
socket = this.#createWebSocket(url.toString());
} catch (error) {
reject(error);
return;
}
let opened = false;
let settled = false;
let callbackFailure = null;
let callbackTail = Promise.resolve();
let ownershipReady = ownership === undefined || ownership === null;
const bufferedEnvelopes = [];
const finish = (error) => {
if (settled) return;
settled = true;
socket.removeEventListener('open', handleOpen);
socket.removeEventListener('message', handleMessage);
socket.removeEventListener('close', handleClose);
socket.removeEventListener('error', handleError);
signal.removeEventListener('abort', handleAbort);
if (ownership?.reconnect === close) ownership.reconnect = null;
void callbackTail.then(() => {
const failure = error ?? callbackFailure;
if (failure) reject(failure);
else resolve();
}, reject);
};
const close = () => {
try {
if (socket.readyState === 0 || socket.readyState === 1) socket.close();
} catch {
// Cleanup must still settle the watcher if a WebSocket rejects close while connecting.
}
};
const handleOpen = () => {
opened = true;
if (ownership) ownership.reconnect = close;
try {
onOpen?.();
} catch (error) {
console.warn(`[${this.#logPrefix}] ignored an interaction open callback failure:`, error.message);
}
if (ownership) {
void this.#refreshInteractionOwnerships(sessionId, signal).then(() => {
if (settled) return;
ownershipReady = true;
for (const envelope of bufferedEnvelopes.splice(0)) processEnvelope(envelope);
}).catch((error) => {
callbackFailure ??= error;
close();
finish(error);
});
}
};
const dispatch = (callback, value) => {
if (!callback) return;
callbackTail = callbackTail
.then(() => callback(value))
.catch((error) => {
callbackFailure ??= error;
close();
finish(callbackFailure);
});
};
const processEnvelope = (envelope) => {
const payload = envelope.payload;
if (ownership && payload.type === 'session/event') {
this.#consumeInteractionOwnerships(sessionId, [payload.event]);
return;
}
if (payload.type === 'question/requested' || payload.type === 'approval/requested') {
const kind = payload.type === 'question/requested' ? 'question' : 'approval';
const interactionId = kind === 'question' ? envelope.rpcId : payload.approvalId;
const claimKey = `${kind}:${interactionId}`;
if (ownership) {
const claim = this.#interactionOwner(sessionId, claimKey, kind);
if (claim?.ownership !== ownership) return;
this.#interactionClaims.set(claimKey, claim);
}
dispatch(onInteraction, Object.freeze({
kind,
interactionId,
rpcId: envelope.rpcId,
sessionId,
payload,
recovered: ownership
? this.#interactionClaims.get(claimKey)?.recovered === true
: false,
reconnect: close,
respond: (result, options = {}) => this.respondInteraction(
envelope.rpcId,
result,
{ ...options, signal: options.signal ?? signal },
),
}));
return;
}
if (payload.type === 'question/resolved' || payload.type === 'approval/resolved') {
const kind = payload.type === 'question/resolved' ? 'question' : 'approval';
const interactionId = kind === 'question'
? payload.questionRpcId
: payload.approvalId;
const claimKey = `${kind}:${interactionId}`;
if (ownership) {
const claim = this.#interactionClaims.get(claimKey);
if (claim?.ownership !== ownership) return;
this.#interactionClaims.delete(claimKey);
}
dispatch(onResolved, Object.freeze({
kind,
interactionId,
sessionId,
outcome: payload.outcome,
payload,
}));
}
};
const handleMessage = (event) => {
try {
if (typeof event.data !== 'string') throw new Error('binary WebSocket frame');
const envelope = JSON.parse(event.data);
const payload = envelope?.payload;
if (envelope?.type !== 'server-request'
|| typeof envelope.rpcId !== 'string'
|| !payload || typeof payload !== 'object'
|| envelope.method !== payload.type) {
throw new Error('invalid server-request envelope');
}
if (payload.sessionId !== sessionId) return;
if (!ownershipReady) bufferedEnvelopes.push(envelope);
else processEnvelope(envelope);
} catch (error) {
console.warn(`[${this.#logPrefix}] ignored a malformed Harness interaction frame:`, error.message);
}
};
const handleClose = () => finish(opened ? null : new Error(
'Harness interaction WebSocket closed before opening',
));
const handleError = () => {
finish(new Error(opened
? 'Harness interaction WebSocket failed'
: 'Harness interaction WebSocket failed before opening'));
close();
};
const handleAbort = () => {
close();
finish();
};
socket.addEventListener('open', handleOpen);
socket.addEventListener('message', handleMessage);
socket.addEventListener('close', handleClose, { once: true });
socket.addEventListener('error', handleError, { once: true });
signal.addEventListener('abort', handleAbort, { once: true });
if (signal.aborted) handleAbort();
});
}
stopManagedProcess() {
if (this.#managedProcess?.exitCode === null) this.#managedProcess.kill('SIGTERM');
}
}

View file

@ -0,0 +1,85 @@
function nonEmptyString(value) {
return typeof value === 'string' && value.trim() ? value.trim() : null;
}
export function validHarnessQuestion(question) {
return question && typeof question.id === 'string' && typeof question.question === 'string'
&& (question.header === undefined || typeof question.header === 'string')
&& (question.detail === undefined || typeof question.detail === 'string')
&& (question.multiSelect === undefined || typeof question.multiSelect === 'boolean')
&& (question.options === undefined || (Array.isArray(question.options)
&& question.options.every((option) => (
option && typeof option.label === 'string'
&& (option.description === undefined || typeof option.description === 'string')
))));
}
export function harnessQuestionText(question, index, total, { requiresMention = false } = {}) {
const lines = [];
const progress = total > 1 ? `(${index + 1}/${total})` : '';
lines.push(`DeepSeek Harness 需要你补充信息${progress}:`);
if (nonEmptyString(question.header)) lines.push('', question.header.trim());
lines.push('', nonEmptyString(question.question) ?? '请输入你的回答。');
if (nonEmptyString(question.detail)) lines.push('', question.detail.trim());
const options = Array.isArray(question.options) ? question.options : [];
if (options.length > 0) {
lines.push('');
options.forEach((option, optionIndex) => {
const label = typeof option?.label === 'string' ? option.label : '';
const description = nonEmptyString(option?.description);
lines.push(`${optionIndex + 1}. ${label}${description ? ` — ${description}` : ''}`);
});
lines.push('', question.multiSelect === true
? '请回复选项序号或文字;多选用逗号分隔,也可补充其他内容。'
: '请回复一个选项序号或文字,也可直接输入其他答案。');
} else {
lines.push('', '请直接回复你的答案。');
}
if (requiresMention) lines.push('', '群聊中请 @机器人 后发送答案。');
return lines.join('\n');
}
function optionLabel(token, options) {
const normalized = token.trim();
if (!normalized) return null;
if (/^\d+$/.test(normalized)) {
const option = options[Number(normalized) - 1];
return typeof option?.label === 'string' ? option.label : null;
}
const exact = options.find((option) => option?.label === normalized);
return typeof exact?.label === 'string' ? exact.label : null;
}
export function harnessAnswerForQuestion(question, text) {
const options = Array.isArray(question.options) ? question.options : [];
if (options.length === 0) {
return { id: question.id, selected: [], custom: text };
}
const wholeLabel = optionLabel(text, options);
if (question.multiSelect !== true) {
return wholeLabel
? { id: question.id, selected: [wholeLabel] }
: { id: question.id, selected: [], custom: text };
}
if (wholeLabel) return { id: question.id, selected: [wholeLabel] };
const selected = [];
const custom = [];
for (const token of text.split(/[,,、;;\n]+/)) {
const value = token.trim();
if (!value) continue;
const label = optionLabel(value, options);
if (label) {
if (!selected.includes(label)) selected.push(label);
} else {
custom.push(value);
}
}
return {
id: question.id,
selected,
...(custom.length > 0 ? { custom: custom.join('、') } : {}),
};
}

View file

@ -6,6 +6,25 @@ import {
DingtalkHarnessBridge,
} from '../../../src/channels/dingtalk/dingtalk-bridge.mjs';
function deferred() {
let resolve;
let reject;
const promise = new Promise((promiseResolve, promiseReject) => {
resolve = promiseResolve;
reject = promiseReject;
});
return { promise, resolve, reject };
}
async function eventually(predicate, message = 'condition was not met') {
const deadline = Date.now() + 1_000;
while (Date.now() < deadline) {
if (predicate()) return;
await new Promise((resolve) => setTimeout(resolve, 5));
}
assert.fail(message);
}
function message(id, text, overrides = {}) {
return {
msgId: id,
@ -269,3 +288,826 @@ test('commands stay local and unsafe session webhooks are rejected before Harnes
assert.equal(sent.length, 1);
assert.equal(bridge.status.lastError, '钉钉消息没有安全的回复地址。');
});
test('a DingTalk reply answers a pending Harness question before the original turn queue', async () => {
const fixture = stateFixture();
const sent = [];
const asked = [];
const response = deferred();
const releaseTurn = deferred();
const bridge = new DingtalkHarnessBridge({
api: { sendText: async (request) => sent.push(request) },
clientId: 'ding-client',
clientSecret: 'host-secret',
harness: {
sessionExists: async () => false,
createSession: async () => 'session-question',
ask: async (sessionId, text, options) => {
asked.push({ sessionId, text });
await options.onInteraction({
kind: 'question',
rpcId: 'question-rpc',
sessionId,
payload: {
type: 'question/requested',
sessionId,
questions: [{
id: 'environment',
header: '运行环境',
question: '请问要使用哪个环境?',
options: [
{ label: '生产环境', description: '处理真实数据' },
{ label: '预发环境' },
],
}],
},
respond: async (result) => {
response.resolve(result);
return { accepted: true };
},
});
await response.promise;
await releaseTurn.promise;
return '已按测试环境继续。';
},
},
state: fixture.state,
});
let firstSettled = false;
const first = bridge.accept(message('prompt', '先问我一个问题'))
.finally(() => { firstSettled = true; });
await eventually(() => sent.some(({ text }) => text.includes('请问要使用哪个环境')));
await Promise.race([
bridge.accept(message('answer', '测试环境')),
new Promise((_, reject) => setTimeout(() => reject(new Error('question reply deadlocked')), 500)),
]);
assert.equal(firstSettled, false);
assert.deepEqual(await response.promise, {
ok: true,
value: {
sessionId: 'session-question',
answer: {
answers: [{ id: 'environment', selected: [], custom: '测试环境' }],
},
},
});
assert.deepEqual(asked, [{ sessionId: 'session-question', text: '先问我一个问题' }]);
releaseTurn.resolve();
await first;
assert.equal(sent.length, 2);
assert.match(sent[0].text, /DeepSeek Harness 需要你补充信息/);
assert.equal(sent[1].text, '已按测试环境继续。');
assert.equal(bridge.status.messagesReceived, 2);
assert.equal(bridge.status.messagesReplied, 1);
});
test('pending questions are isolated by DingTalk conversation', async () => {
const fixture = stateFixture();
fixture.sessions.set('p2p:staff-a', 'session-a');
fixture.sessions.set('p2p:staff-b', 'session-b');
const sent = [];
const asked = [];
const answeredA = deferred();
const releaseA = deferred();
const bridge = new DingtalkHarnessBridge({
api: { sendText: async (request) => sent.push(request) },
clientId: 'ding-client',
clientSecret: 'host-secret',
harness: {
sessionExists: async () => true,
createSession: async () => assert.fail('existing sessions should be reused'),
ask: async (sessionId, text, options) => {
asked.push({ sessionId, text });
if (sessionId === 'session-b') return '乙会话的普通回答';
await options.onInteraction({
kind: 'question',
rpcId: 'rpc-a',
sessionId,
payload: {
type: 'question/requested',
sessionId,
questions: [{ id: 'a', question: '甲会话的问题' }],
},
respond: async (result) => {
answeredA.resolve(result);
return { accepted: true };
},
});
await answeredA.promise;
await releaseA.promise;
return '甲会话完成';
},
},
state: fixture.state,
});
const firstA = bridge.accept(message('a-prompt', '启动甲会话', {
senderStaffId: 'staff-a',
}));
await eventually(() => sent.some(({ text }) => text.includes('甲会话的问题')));
await bridge.accept(message('b-message', '乙会话的消息', {
senderStaffId: 'staff-b',
}));
assert.deepEqual(asked, [
{ sessionId: 'session-a', text: '启动甲会话' },
{ sessionId: 'session-b', text: '乙会话的消息' },
]);
assert.equal(sent.some(({ text }) => text === '乙会话的普通回答'), true);
await bridge.accept(message('a-answer', '甲的答案', { senderStaffId: 'staff-a' }));
assert.deepEqual((await answeredA.promise).value.answer.answers, [
{ id: 'a', selected: [], custom: '甲的答案' },
]);
releaseA.resolve();
await firstA;
});
test('question replays are deduplicated and approval interactions are never auto-approved', async () => {
const fixture = stateFixture();
const sent = [];
let approvalResponses = 0;
let secondQuestionResponse;
const bridge = new DingtalkHarnessBridge({
api: { sendText: async (request) => sent.push(request) },
clientId: 'ding-client',
clientSecret: 'host-secret',
harness: {
sessionExists: async () => false,
createSession: async () => 'session-replay',
ask: async (sessionId, _text, options) => {
const question = {
kind: 'question',
rpcId: 'replayed-question',
sessionId,
payload: {
type: 'question/requested',
sessionId,
questions: [{ id: 'choice', question: '只应显示一次' }],
},
respond: async () => ({ accepted: true }),
};
await options.onInteraction(question);
await options.onInteraction(question);
await options.onInteraction({
kind: 'question',
interactionId: 'parallel-question',
rpcId: 'parallel-question',
sessionId,
payload: {
type: 'question/requested',
sessionId,
questions: [{ id: 'parallel', question: '不应无声丢弃' }],
},
respond: async (result) => {
secondQuestionResponse = result;
return { accepted: true };
},
});
await options.onInteraction({
kind: 'approval',
rpcId: 'approval-rpc',
sessionId,
payload: {
type: 'approval/requested',
sessionId,
approvalId: 'approval-one',
toolName: 'bash',
},
respond: async () => { approvalResponses += 1; },
});
await options.onInteractionResolved({
kind: 'question',
sessionId,
interactionId: 'replayed-question',
outcome: 'cancelled',
});
return '交互已取消';
},
},
state: fixture.state,
logger: { warn() {}, error() {} },
});
await bridge.accept(message('replay', '测试重放'));
assert.equal(sent.filter(({ text }) => text.includes('只应显示一次')).length, 1);
assert.deepEqual(secondQuestionResponse, {
ok: false,
error: {
code: 'cancelled',
message: 'DingTalk is already handling another user interaction.',
details: {},
},
});
assert.equal(sent.some(({ text }) => text.includes('approval')), false);
assert.equal(approvalResponses, 0);
assert.equal(sent.at(-1).text, '交互已取消');
});
test('a message arriving during question submission is re-queued as the next normal prompt', async () => {
const fixture = stateFixture();
const sent = [];
const asked = [];
const submitStarted = deferred();
const releaseSubmit = deferred();
const answered = deferred();
const bridge = new DingtalkHarnessBridge({
api: { sendText: async (request) => sent.push(request) },
clientId: 'ding-client',
clientSecret: 'host-secret',
harness: {
sessionExists: async () => true,
createSession: async () => 'session-submit-race',
ask: async (sessionId, text, options) => {
asked.push(text);
if (text === '下一个真实问题') return '第二轮回答';
const interaction = {
kind: 'question',
interactionId: 'submit-race-question',
rpcId: 'submit-race-question',
sessionId,
payload: {
type: 'question/requested',
sessionId,
questions: [{ id: 'answer', question: '请先回答当前问题' }],
},
respond: async (result) => {
submitStarted.resolve(result);
await releaseSubmit.promise;
answered.resolve();
return { accepted: true };
},
};
await options.onInteraction(interaction);
await answered.promise;
return '第一轮回答';
},
},
state: fixture.state,
});
const first = bridge.accept(message('submit-start', '启动交互'));
await eventually(() => sent.some(({ text }) => text.includes('请先回答当前问题')));
const answer = bridge.accept(message('submit-answer', '当前问题的答案'));
await submitStarted.promise;
let nextSettled = false;
const next = bridge.accept(message('submit-next', '下一个真实问题'))
.finally(() => { nextSettled = true; });
await new Promise((resolve) => setTimeout(resolve, 10));
assert.equal(nextSettled, false);
releaseSubmit.resolve();
await Promise.all([answer, first, next]);
assert.deepEqual(asked, ['启动交互', '下一个真实问题']);
assert.equal(asked.includes('当前问题的答案'), false);
assert.deepEqual(sent.slice(-2).map(({ text }) => text), ['第一轮回答', '第二轮回答']);
});
test('a queued next prompt stays separate when question submission must be retried', async () => {
const fixture = stateFixture();
const sent = [];
const asked = [];
const firstSubmitStarted = deferred();
const releaseFirstSubmit = deferred();
const answered = deferred();
const submittedAnswers = [];
let submitAttempts = 0;
const bridge = new DingtalkHarnessBridge({
api: { sendText: async (request) => sent.push(request) },
clientId: 'ding-client',
clientSecret: 'host-secret',
harness: {
sessionExists: async () => true,
createSession: async () => 'session-submit-retry-race',
ask: async (sessionId, text, options) => {
asked.push(text);
if (text === '排队的下一个问题') return '第二轮完成';
await options.onInteraction({
kind: 'question',
interactionId: 'submit-retry-question',
rpcId: 'submit-retry-question',
sessionId,
payload: {
type: 'question/requested',
sessionId,
questions: [{ id: 'answer', question: '请回答后再继续' }],
},
respond: async (result) => {
submittedAnswers.push(result.value.answer.answers[0].custom);
submitAttempts += 1;
if (submitAttempts === 1) {
firstSubmitStarted.resolve();
await releaseFirstSubmit.promise;
throw new Error('temporary response failure');
}
answered.resolve();
return { accepted: true };
},
});
await answered.promise;
return '第一轮完成';
},
},
state: fixture.state,
logger: { error() {}, warn() {} },
});
const first = bridge.accept(message('submit-retry-start', '启动可重试交互'));
await eventually(() => sent.some(({ text }) => text.includes('请回答后再继续')));
const firstAnswer = bridge.accept(message('submit-retry-answer', '第一次答案'));
await firstSubmitStarted.promise;
let nextSettled = false;
const next = bridge.accept(message('submit-retry-next', '排队的下一个问题'))
.finally(() => { nextSettled = true; });
releaseFirstSubmit.resolve();
await firstAnswer;
await eventually(() => sent.some(({ text }) => text.includes('回答提交失败')));
assert.equal(nextSettled, false);
assert.deepEqual(asked, ['启动可重试交互']);
const retry = bridge.accept(message('submit-retry-again', '重试后的答案'));
await Promise.all([retry, first, next]);
assert.deepEqual(submittedAnswers, ['第一次答案', '重试后的答案']);
assert.deepEqual(asked, ['启动可重试交互', '排队的下一个问题']);
assert.deepEqual(sent.slice(-2).map(({ text }) => text), ['第一轮完成', '第二轮完成']);
});
test('an invalid pending reply does not block the valid answer behind it', async () => {
const fixture = stateFixture();
const sent = [];
const invalidNoticeStarted = deferred();
const releaseInvalidNotice = deferred();
const answered = deferred();
let submitted;
const bridge = new DingtalkHarnessBridge({
api: {
sendText: async (request) => {
if (request.text === '请用文字回答当前问题。') {
invalidNoticeStarted.resolve();
await releaseInvalidNotice.promise;
}
sent.push(request);
},
},
clientId: 'ding-client',
clientSecret: 'host-secret',
harness: {
sessionExists: async () => true,
createSession: async () => 'session-invalid-reply',
ask: async (sessionId, _text, options) => {
await options.onInteraction({
kind: 'question',
interactionId: 'invalid-reply-question',
rpcId: 'invalid-reply-question',
sessionId,
payload: {
type: 'question/requested',
sessionId,
questions: [{ id: 'answer', question: '请给出有效文字答案' }],
},
respond: async (result) => {
submitted = result;
answered.resolve();
return { accepted: true };
},
});
await answered.promise;
return '有效答案已收到';
},
},
state: fixture.state,
});
const first = bridge.accept(message('invalid-reply-start', '启动交互'));
await eventually(() => sent.some(({ text }) => text.includes('请给出有效文字答案')));
const invalid = bridge.accept(message('invalid-reply-image', '', {
msgtype: 'picture',
text: undefined,
}));
await invalidNoticeStarted.promise;
const valid = bridge.accept(message('invalid-reply-valid', '真正的答案'));
releaseInvalidNotice.resolve();
await Promise.all([invalid, valid, first]);
assert.deepEqual(submitted.value.answer.answers, [{
id: 'answer',
selected: [],
custom: '真正的答案',
}]);
assert.equal(sent.at(-1).text, '有效答案已收到');
});
test('an answer resolved elsewhere is not reinterpreted as a later prompt', async () => {
const fixture = stateFixture();
const originalMarkSeen = fixture.state.markSeen;
const answerMarkStarted = deferred();
const releaseAnswerMark = deferred();
fixture.state.markSeen = async (id) => {
if (id === 'resolved-answer-first') {
answerMarkStarted.resolve();
await releaseAnswerMark.promise;
}
await originalMarkSeen(id);
};
const sent = [];
const asked = [];
const resolved = deferred();
let resolveInteraction;
const bridge = new DingtalkHarnessBridge({
api: { sendText: async (request) => sent.push(request) },
clientId: 'ding-client',
clientSecret: 'host-secret',
harness: {
sessionExists: async () => true,
createSession: async () => 'session-resolved-race',
ask: async (sessionId, text, options) => {
asked.push(text);
if (text === '后来的普通问题') return '后来问题的回答';
await options.onInteraction({
kind: 'question',
interactionId: 'resolved-race-question',
rpcId: 'resolved-race-question',
sessionId,
payload: {
type: 'question/requested',
sessionId,
questions: [{ id: 'answer', question: '可能在其他客户端回答' }],
},
respond: async () => ({ accepted: true }),
});
resolveInteraction = async () => {
await options.onInteractionResolved({
kind: 'question',
interactionId: 'resolved-race-question',
sessionId,
outcome: 'answered',
});
resolved.resolve();
};
await resolved.promise;
return '第一轮已由其他客户端完成';
},
},
state: fixture.state,
});
const first = bridge.accept(message('resolved-race-start', '启动外部解决竞态'));
await eventually(() => typeof resolveInteraction === 'function');
const answer = bridge.accept(message('resolved-answer-first', '原本的问题答案'));
await answerMarkStarted.promise;
const later = bridge.accept(message('resolved-later-second', '后来的普通问题'));
await resolveInteraction();
releaseAnswerMark.resolve();
await Promise.all([answer, first, later]);
assert.deepEqual(asked, ['启动外部解决竞态', '后来的普通问题']);
assert.equal(asked.includes('原本的问题答案'), false);
assert.equal(sent.some(({ text }) => text.includes('已在其他客户端处理')), true);
});
test('an orphan question is cancelled without exposing it to a new conversation', async () => {
const fixture = stateFixture();
const sent = [];
let recoveredResponse;
const bridge = new DingtalkHarnessBridge({
api: { sendText: async (request) => sent.push(request) },
clientId: 'ding-client',
clientSecret: 'host-secret',
harness: {
sessionExists: async () => true,
createSession: async () => 'session-orphan-recovery',
ask: async (sessionId, _text, options) => {
await options.onInteraction({
kind: 'question',
interactionId: 'orphan-secret-question',
rpcId: 'orphan-secret-question',
sessionId,
recovered: true,
payload: {
type: 'question/requested',
sessionId,
questions: [{ id: 'secret', question: '旧会话中的敏感问题内容' }],
},
respond: async (result) => {
recoveredResponse = result;
return { accepted: true };
},
});
return '新的消息已继续';
},
},
state: fixture.state,
});
await bridge.accept(message('orphan-recovery', '新的会话消息'));
assert.deepEqual(recoveredResponse, {
ok: false,
error: {
code: 'cancelled',
message: 'DingTalk safely cancelled an interaction left by an earlier client.',
details: {},
},
});
assert.equal(sent.some(({ text }) => text.includes('旧会话中的敏感问题内容')), false);
assert.equal(sent.some(({ text }) => text.includes('遗留的待回答问题')), true);
assert.equal(sent.at(-1).text, '新的消息已继续');
});
test('question delivery replays safely and a multi-question batch keeps canonical answers', async () => {
const fixture = stateFixture();
const sent = [];
const response = deferred();
let firstQuestionAttempts = 0;
let secondQuestionAttempts = 0;
let replayTask = Promise.resolve();
const bridge = new DingtalkHarnessBridge({
api: {
sendText: async (request) => {
if (request.text.includes('选择回答语言')) {
firstQuestionAttempts += 1;
if (firstQuestionAttempts === 1) throw new Error('temporary first question failure');
}
if (request.text.includes('选择交付内容')) {
secondQuestionAttempts += 1;
if (secondQuestionAttempts === 1) throw new Error('temporary second question failure');
}
sent.push(request);
},
},
clientId: 'ding-client',
clientSecret: 'host-secret',
harness: {
sessionExists: async () => false,
createSession: async () => 'session-question-batch',
ask: async (sessionId, _text, options) => {
// An already-started turn remains the owner even if a workspace switch
// clears the conversation-to-session mapping while it is running.
fixture.sessions.delete('p2p:staff-approved');
const interaction = {
kind: 'question',
interactionId: 'batch-question',
rpcId: 'batch-question',
sessionId,
payload: {
type: 'question/requested',
sessionId,
questions: [
{
id: 'language',
question: '选择回答语言',
options: [{ label: '中文' }, { label: 'English' }],
},
{
id: 'deliverables',
question: '选择交付内容',
multiSelect: true,
options: [{ label: '测试' }, { label: '文档' }],
},
],
},
reconnect: () => {
replayTask = new Promise((resolve, reject) => {
queueMicrotask(() => options.onInteraction(interaction).then(resolve, reject));
});
},
respond: async (result) => {
response.resolve(result);
return { accepted: true };
},
};
await assert.rejects(
options.onInteraction(interaction),
/temporary first question failure/,
);
await options.onInteraction(interaction);
await response.promise;
return '批量问题已完成';
},
},
state: fixture.state,
logger: { error() {}, warn() {} },
});
const first = bridge.accept(message('batch-start', '请分步提问'));
await eventually(() => sent.some(({ text }) => text.includes('选择回答语言')));
await bridge.accept(message('batch-language', '2'));
await eventually(() => sent.some(({ text }) => text.includes('选择交付内容')));
await replayTask;
await bridge.accept(message('batch-deliverables', '1,文档,发布说明'));
assert.deepEqual(await response.promise, {
ok: true,
value: {
sessionId: 'session-question-batch',
answer: {
answers: [
{ id: 'language', selected: ['English'] },
{ id: 'deliverables', selected: ['测试', '文档'], custom: '发布说明' },
],
},
},
});
assert.equal(firstQuestionAttempts, 2);
assert.equal(secondQuestionAttempts, 2);
await first;
assert.equal(sent.at(-1).text, '批量问题已完成');
});
test('aborting a DingTalk turn cancels its pending Harness question', async () => {
const fixture = stateFixture();
const controller = new AbortController();
const sent = [];
const cancelled = deferred();
const bridge = new DingtalkHarnessBridge({
api: { sendText: async (request) => sent.push(request) },
clientId: 'ding-client',
clientSecret: 'host-secret',
harness: {
sessionExists: async () => false,
createSession: async () => 'session-cancel-question',
ask: async (sessionId, _text, options) => {
await options.onInteraction({
kind: 'question',
interactionId: 'question-to-cancel',
rpcId: 'question-to-cancel',
sessionId,
payload: {
type: 'question/requested',
sessionId,
questions: [{ id: 'cancel', question: '等待重连前取消' }],
},
respond: async (result) => {
cancelled.resolve(result);
return { accepted: true };
},
});
await new Promise((resolve, reject) => {
if (options.signal.aborted) {
reject(options.signal.reason);
return;
}
options.signal.addEventListener('abort', () => reject(options.signal.reason), {
once: true,
});
});
},
},
state: fixture.state,
signal: controller.signal,
});
const processing = bridge.accept(message('cancel-start', '启动后停止'));
await eventually(() => sent.some(({ text }) => text.includes('等待重连前取消')));
controller.abort(new DOMException('runtime stopped', 'AbortError'));
await processing;
assert.deepEqual(await cancelled.promise, {
ok: false,
error: {
code: 'cancelled',
message: 'The DingTalk interaction ended before the user answered.',
details: {},
},
});
});
test('group questions tell users to mention the bot and ignore unmentioned answers', async () => {
const fixture = stateFixture();
const sent = [];
const answered = deferred();
let responseCalls = 0;
const bridge = new DingtalkHarnessBridge({
api: { sendText: async (request) => sent.push(request) },
clientId: 'ding-client',
clientSecret: 'host-secret',
harness: {
sessionExists: async () => false,
createSession: async () => 'session-group-question',
ask: async (sessionId, _text, options) => {
await options.onInteraction({
kind: 'question',
interactionId: 'group-question',
rpcId: 'group-question',
sessionId,
payload: {
type: 'question/requested',
sessionId,
questions: [{ id: 'group', question: '群聊问题' }],
},
respond: async () => {
responseCalls += 1;
answered.resolve();
return { accepted: true };
},
});
await answered.promise;
return '群聊交互完成';
},
},
state: fixture.state,
});
const group = {
conversationType: '2',
conversationId: 'group-question-room',
};
const first = bridge.accept(message('group-start', '开始群聊问答', {
...group,
isInAtList: true,
}));
await eventually(() => sent.some(({ text }) => text.includes('群聊问题')));
assert.match(sent[0].text, /群聊中请 @机器人/);
await bridge.accept(message('group-unmentioned', '这条不算答案', {
...group,
isInAtList: false,
}));
assert.equal(responseCalls, 0);
await bridge.accept(message('group-mentioned', '这条才是答案', {
...group,
isInAtList: true,
}));
await first;
assert.equal(responseCalls, 1);
assert.equal(bridge.status.messagesIgnored, 1);
assert.equal(sent.at(-1).text, '群聊交互完成');
});
test('only the actor who started a DingTalk group interaction can answer it', async () => {
const fixture = stateFixture();
fixture.sessions.set('group:actor-room', 'session-group-actor');
const asked = [];
const submitted = deferred();
let interactionSubmitted = false;
const bridge = new DingtalkHarnessBridge({
api: { sendText: async () => undefined },
clientId: 'ding-client',
clientSecret: 'host-secret',
harness: {
sessionExists: async () => true,
createSession: async () => assert.fail('the group session should already exist'),
ask: async (sessionId, text, options) => {
asked.push(text);
if (text !== '甲发起交互') return '普通群消息已处理';
await options.onInteraction({
kind: 'question',
interactionId: 'dingtalk-group-actor',
rpcId: 'dingtalk-group-actor',
sessionId,
payload: {
type: 'question/requested',
sessionId,
questions: [{ id: 'actor', question: '只能由甲回答' }],
},
respond: async (result) => {
interactionSubmitted = true;
submitted.resolve(result);
return { accepted: true };
},
});
await submitted.promise;
return '甲的交互已完成';
},
},
state: fixture.state,
});
const group = {
conversationType: '2',
conversationId: 'actor-room',
isInAtList: true,
};
const first = bridge.accept(message('actor-start', '甲发起交互', {
...group,
senderStaffId: 'staff-a',
}));
await eventually(() => asked.length === 1);
const intruder = bridge.accept(message('actor-b', '乙试图代答', {
...group,
senderStaffId: 'staff-b',
}));
await new Promise((resolve) => setTimeout(resolve, 20));
assert.equal(interactionSubmitted, false);
assert.deepEqual(asked, ['甲发起交互']);
await bridge.accept(message('actor-a', '甲的答案', {
...group,
senderStaffId: 'staff-a',
}));
assert.deepEqual((await submitted.promise).value.answer.answers, [{
id: 'actor',
selected: [],
custom: '甲的答案',
}]);
await Promise.all([first, intruder]);
assert.deepEqual(asked, ['甲发起交互', '乙试图代答']);
});

View file

@ -1,7 +1,69 @@
import assert from 'node:assert/strict';
import test from 'node:test';
import { HarnessClient, HarnessReplyTracker, HarnessRpcError } from '../../../src/channels/dingtalk/harness-client.mjs';
import {
HarnessClient,
HarnessInteractionError,
HarnessReplyTracker,
HarnessRpcError,
} from '../../../src/channels/dingtalk/harness-client.mjs';
import {
HarnessClient as FeishuHarnessClient,
} from '../../../src/channels/feishu/harness-client.mjs';
function deferred() {
let resolve;
let reject;
const promise = new Promise((promiseResolve, promiseReject) => {
resolve = promiseResolve;
reject = promiseReject;
});
return { promise, resolve, reject };
}
async function eventually(predicate, message = 'condition was not met') {
const deadline = Date.now() + 1_000;
while (Date.now() < deadline) {
if (predicate()) return;
await new Promise((resolve) => setTimeout(resolve, 5));
}
assert.fail(message);
}
class FakeSocket {
#listeners = new Map();
readyState = 0;
addEventListener(name, listener) {
const listeners = this.#listeners.get(name) ?? new Set();
listeners.add(listener);
this.#listeners.set(name, listeners);
}
removeEventListener(name, listener) {
this.#listeners.get(name)?.delete(listener);
}
open() {
if (this.readyState !== 0) return;
this.readyState = 1;
this.emit('open', {});
}
frame(value) {
this.emit('message', { data: JSON.stringify(value) });
}
close(code = 1000) {
if (this.readyState >= 2) return;
this.readyState = 3;
this.emit('close', { code });
}
emit(name, event) {
for (const listener of [...(this.#listeners.get(name) ?? [])]) listener(event);
}
}
test('HarnessClient lists only absolute workspace paths and forwards request options', async () => {
const client = new HarnessClient({
@ -391,3 +453,818 @@ test('Harness client validates the RPC envelope and preserves server error codes
assert.equal(request.url, 'http://127.0.0.1:3080/api/session.history');
assert.match(request.body.rpcId, /^dingtalk-/);
});
test('interaction watcher decodes mux requests, filters sessions, and posts exact responses', async () => {
const requests = [];
const opened = deferred();
let socket;
let socketUrl;
const client = new HarnessClient({
baseUrl: 'http://127.0.0.1:3080/base',
workspace: '/tmp/workspace',
fetchImpl: async (url, options) => {
requests.push({
url: url.toString(),
method: options.method,
body: JSON.parse(options.body),
});
return { ok: true, json: async () => ({ accepted: true }) };
},
createWebSocket: (url) => {
socketUrl = url;
socket = new FakeSocket();
queueMicrotask(() => socket.open());
return socket;
},
});
const controller = new AbortController();
const interactions = [];
const resolutions = [];
const watching = client.watchInteractions('session-one', {
signal: controller.signal,
onOpen: opened.resolve,
onInteraction: (interaction) => interactions.push(interaction),
onResolved: (resolution) => resolutions.push(resolution),
});
await opened.promise;
socket.frame({
type: 'server-request',
rpcId: 'other-rpc',
method: 'question/requested',
payload: {
type: 'question/requested',
sessionId: 'session-other',
questions: [{ id: 'ignored', question: 'ignore me' }],
},
});
socket.frame({
type: 'server-request',
rpcId: 'approval-rpc',
method: 'approval/requested',
payload: {
type: 'approval/requested',
sessionId: 'session-one',
approvalId: 'approval-one',
toolName: 'bash',
},
});
socket.frame({
type: 'server-request',
rpcId: 'question-rpc',
method: 'question/requested',
payload: {
type: 'question/requested',
sessionId: 'session-one',
questions: [{ id: 'language', question: 'Which language?' }],
},
});
socket.frame({
type: 'server-request',
rpcId: 'resolution-envelope',
method: 'question/resolved',
payload: {
type: 'question/resolved',
sessionId: 'session-one',
questionRpcId: 'question-rpc',
outcome: 'answered',
},
});
socket.frame({
type: 'server-request',
rpcId: 'approval-resolution-envelope',
method: 'approval/resolved',
payload: {
type: 'approval/resolved',
sessionId: 'session-one',
approvalId: 'approval-one',
outcome: 'rejected',
},
});
await eventually(() => interactions.length === 2 && resolutions.length === 2);
assert.equal(socketUrl, 'ws://127.0.0.1:3080/api/events.mux');
assert.equal(interactions.length, 2);
assert.deepEqual({
kind: interactions[0].kind,
interactionId: interactions[0].interactionId,
rpcId: interactions[0].rpcId,
sessionId: interactions[0].sessionId,
}, {
kind: 'approval',
interactionId: 'approval-one',
rpcId: 'approval-rpc',
sessionId: 'session-one',
});
assert.deepEqual({
kind: interactions[1].kind,
interactionId: interactions[1].interactionId,
rpcId: interactions[1].rpcId,
sessionId: interactions[1].sessionId,
}, {
kind: 'question',
interactionId: 'question-rpc',
rpcId: 'question-rpc',
sessionId: 'session-one',
});
assert.deepEqual(resolutions.map(({ kind, interactionId, outcome }) => ({
kind,
interactionId,
outcome,
})), [
{
kind: 'question',
interactionId: 'question-rpc',
outcome: 'answered',
},
{
kind: 'approval',
interactionId: 'approval-one',
outcome: 'rejected',
},
]);
assert.equal(requests.length, 0, 'receiving an approval must not approve it automatically');
const result = {
ok: true,
value: {
sessionId: 'session-one',
answer: { answers: [{ id: 'language', selected: [], custom: 'Chinese' }] },
},
};
assert.deepEqual(await interactions[1].respond(result), { accepted: true });
assert.deepEqual(requests, [{
url: 'http://127.0.0.1:3080/api/respond',
method: 'POST',
body: { type: 'client-response', rpcId: 'question-rpc', result },
}]);
controller.abort();
await watching;
assert.equal(socket.readyState, 3);
});
test('interaction watcher reconnects over wss and exposes replayed stable request ids', async () => {
const sockets = [];
const seen = [];
const controller = new AbortController();
const client = new HarnessClient({
baseUrl: 'https://harness.example/nested',
workspace: '/tmp/workspace',
interactionReconnectDelayMs: 0,
createWebSocket: (url) => {
const socket = new FakeSocket();
sockets.push({ socket, url });
queueMicrotask(() => socket.open());
return socket;
},
});
const watching = client.watchInteractions('session-replay', {
signal: controller.signal,
onInteraction: ({ rpcId }) => seen.push(rpcId),
});
await eventually(() => sockets[0]?.socket.readyState === 1);
const frame = {
type: 'server-request',
rpcId: 'stable-question-rpc',
method: 'question/requested',
payload: {
type: 'question/requested',
sessionId: 'session-replay',
questions: [{ id: 'replay', question: 'Still pending?' }],
},
};
sockets[0].socket.frame(frame);
sockets[0].socket.close(1006);
await eventually(() => sockets[1]?.socket.readyState === 1, 'watcher did not reconnect');
sockets[1].socket.frame(frame);
await eventually(() => seen.length === 2);
assert.deepEqual(seen, ['stable-question-rpc', 'stable-question-rpc']);
assert.deepEqual(sockets.map(({ url }) => url), [
'wss://harness.example/api/events.mux',
'wss://harness.example/api/events.mux',
]);
controller.abort();
await watching;
});
test('interaction response receipts distinguish protocol rejection from invalid transport data', async () => {
let receipt = { accepted: false, reason: 'not-pending' };
const client = new HarnessClient({
baseUrl: 'http://127.0.0.1:3080',
workspace: '/tmp/workspace',
fetchImpl: async () => ({ ok: true, json: async () => receipt }),
});
const result = { ok: true, value: { sessionId: 'session-one' } };
await assert.rejects(
client.respondInteraction('question-rpc', result),
(error) => error instanceof HarnessInteractionError
&& error.code === 'interaction-not-pending',
);
receipt = { accepted: false, reason: 'bad-response' };
await assert.rejects(
client.respondInteraction('question-rpc', result),
(error) => error instanceof HarnessInteractionError
&& error.code === 'interaction-bad-response',
);
receipt = { type: 'server-response', accepted: false };
await assert.rejects(
client.respondInteraction('question-rpc', result),
/invalid interaction response receipt/,
);
});
test('ask opens the interaction watcher before prompting and closes it with the turn', async () => {
let socket;
let promptRpcId;
let historyCalls = 0;
const responses = [];
const interactions = [];
const client = new HarnessClient({
baseUrl: 'http://127.0.0.1:3080',
workspace: '/tmp/workspace',
fetchImpl: async (_url, options) => {
responses.push(JSON.parse(options.body));
return { ok: true, json: async () => ({ accepted: true }) };
},
createWebSocket: () => {
const createdSocket = new FakeSocket();
socket = createdSocket;
queueMicrotask(() => createdSocket.open());
return createdSocket;
},
});
client.ensureRunning = async () => true;
client.rpc = async (method, payload, _timeoutMs, options) => {
if (method === 'session.history') {
historyCalls += 1;
if (historyCalls === 1) return { events: [] };
return {
events: [
{ event: { seq: 1, type: 'turn/start', data: { turn: 1 } } },
{ event: {
seq: 2,
type: 'user/message',
data: { turn: 1, source: { rpcId: promptRpcId } },
} },
{ event: {
seq: 3,
type: 'assistant/message',
data: {
turn: 1,
message: { content: [{ type: 'text', text: '已收到回答' }] },
},
} },
{ event: { seq: 4, type: 'turn/end', data: { turn: 1, reason: 'completed' } } },
],
};
}
assert.equal(method, 'session.prompt');
assert.equal(socket.readyState, 1, 'interaction watcher must open before session.prompt');
assert.equal(payload.sessionId, 'session-one');
promptRpcId = options.rpcId;
socket.frame({
type: 'server-request',
rpcId: 'turn-start-frame',
method: 'session/event',
payload: {
type: 'session/event',
sessionId: 'session-one',
event: { seq: 1, type: 'turn/start', data: { turn: 1 } },
},
});
socket.frame({
type: 'server-request',
rpcId: 'user-message-frame',
method: 'session/event',
payload: {
type: 'session/event',
sessionId: 'session-one',
event: {
seq: 2,
type: 'user/message',
data: { turn: 1, source: { rpcId: promptRpcId } },
},
},
});
socket.frame({
type: 'server-request',
rpcId: 'question-during-ask',
method: 'question/requested',
payload: {
type: 'question/requested',
sessionId: 'session-one',
questions: [{
id: 'ready',
question: 'Ready?',
options: [{ label: 'Yes' }, { label: 'No' }],
}],
},
});
return {};
};
const answer = await client.ask('session-one', '请先提问', {
onInteraction: async (interaction) => {
interactions.push(interaction.rpcId);
await interaction.respond({
ok: true,
value: {
sessionId: interaction.sessionId,
answer: { answers: [{ id: 'ready', selected: ['Yes'] }] },
},
});
},
});
assert.equal(answer, '已收到回答');
assert.deepEqual(interactions, ['question-during-ask']);
assert.deepEqual(responses, [{
type: 'client-response',
rpcId: 'question-during-ask',
result: {
ok: true,
value: {
sessionId: 'session-one',
answer: { answers: [{ id: 'ready', selected: ['Yes'] }] },
},
},
}]);
assert.equal(socket.readyState, 3);
});
test('interaction callbacks preserve frame order and watcher shutdown drains them', async () => {
const opened = deferred();
const releaseRequested = deferred();
const order = [];
let socket;
const client = new HarnessClient({
baseUrl: 'http://127.0.0.1:3080',
workspace: '/tmp/workspace',
createWebSocket: () => {
socket = new FakeSocket();
queueMicrotask(() => socket.open());
return socket;
},
});
const controller = new AbortController();
const watching = client.watchInteractions('session-ordered', {
signal: controller.signal,
onOpen: opened.resolve,
onInteraction: async () => {
order.push('requested:start');
await releaseRequested.promise;
order.push('requested:end');
},
onResolved: () => order.push('resolved'),
});
await opened.promise;
socket.frame({
type: 'server-request',
rpcId: 'ordered-question',
method: 'question/requested',
payload: {
type: 'question/requested',
sessionId: 'session-ordered',
questions: [{ id: 'ordered', question: 'Wait for delivery' }],
},
});
socket.frame({
type: 'server-request',
rpcId: 'ordered-resolution',
method: 'question/resolved',
payload: {
type: 'question/resolved',
sessionId: 'session-ordered',
questionRpcId: 'ordered-question',
outcome: 'cancelled',
},
});
controller.abort();
await eventually(() => order.length > 0);
assert.deepEqual(order, ['requested:start']);
let settled = false;
void watching.finally(() => { settled = true; });
await new Promise((resolve) => setTimeout(resolve, 10));
assert.equal(settled, false);
releaseRequested.resolve();
await watching;
assert.deepEqual(order, ['requested:start', 'requested:end', 'resolved']);
});
test('concurrent asks on one Harness session receive interactions only for their own active turn', async () => {
const sockets = [];
const prompts = [];
const received = { first: [], second: [] };
const client = new HarnessClient({
baseUrl: 'http://127.0.0.1:3080',
workspace: '/tmp/workspace',
createWebSocket: () => {
const socket = new FakeSocket();
sockets.push(socket);
queueMicrotask(() => socket.open());
return socket;
},
});
client.ensureRunning = async () => true;
client.rpc = async (method, payload, _timeoutMs, options) => {
if (method === 'session.history') return { events: [] };
assert.equal(method, 'session.prompt');
prompts.push({ text: payload.content[0].text, rpcId: options.rpcId });
return {};
};
const firstController = new AbortController();
const secondController = new AbortController();
const first = client.ask('shared-session', 'first', {
signal: firstController.signal,
onInteraction: ({ rpcId }) => received.first.push(rpcId),
});
const second = client.ask('shared-session', 'second', {
signal: secondController.signal,
onInteraction: ({ rpcId }) => received.second.push(rpcId),
});
await eventually(() => prompts.length === 2 && sockets.length === 2);
const active = prompts[0];
const frames = [
{
type: 'server-request',
rpcId: 'shared-turn-start',
method: 'session/event',
payload: {
type: 'session/event',
sessionId: 'shared-session',
event: { seq: 1, type: 'turn/start', data: { turn: 1 } },
},
},
{
type: 'server-request',
rpcId: 'shared-user-message',
method: 'session/event',
payload: {
type: 'session/event',
sessionId: 'shared-session',
event: {
seq: 2,
type: 'user/message',
data: { source: { rpcId: active.rpcId } },
},
},
},
{
type: 'server-request',
rpcId: 'shared-plugin-context',
method: 'session/event',
payload: {
type: 'session/event',
sessionId: 'shared-session',
event: {
seq: 3,
type: 'user/message',
data: { source: { kind: 'plugin', pluginId: 'time-context' } },
},
},
},
{
type: 'server-request',
rpcId: 'shared-question',
method: 'question/requested',
payload: {
type: 'question/requested',
sessionId: 'shared-session',
questions: [{ id: 'owner', question: 'Who owns this turn?' }],
},
},
];
for (const frame of frames) {
for (const socket of sockets) socket.frame(frame);
}
await eventually(() => received.first.length + received.second.length === 1);
assert.deepEqual(received[active.text], ['shared-question']);
assert.deepEqual(received[active.text === 'first' ? 'second' : 'first'], []);
firstController.abort();
secondController.abort();
await Promise.allSettled([first, second]);
});
test('channel-specific clients share interaction ownership for the same Harness origin', async () => {
const sockets = [];
const prompts = [];
const received = { dingtalk: [], feishu: [] };
const makeClient = (Client, channel) => {
const client = new Client({
baseUrl: 'http://127.0.0.1:3080',
workspace: '/tmp/workspace',
createWebSocket: () => {
const socket = new FakeSocket();
sockets.push({ channel, socket });
queueMicrotask(() => socket.open());
return socket;
},
});
client.ensureRunning = async () => true;
client.rpc = async (method, payload, _timeoutMs, options) => {
if (method === 'session.history') return { events: [] };
assert.equal(method, 'session.prompt');
prompts.push({ channel, rpcId: options.rpcId, text: payload.content[0].text });
return {};
};
return client;
};
const dingtalk = makeClient(HarnessClient, 'dingtalk');
const feishu = makeClient(FeishuHarnessClient, 'feishu');
const dingtalkController = new AbortController();
const feishuController = new AbortController();
const first = dingtalk.ask('cross-channel-session', 'dingtalk prompt', {
signal: dingtalkController.signal,
onInteraction: ({ rpcId }) => received.dingtalk.push(rpcId),
});
await eventually(() => prompts.length === 1);
const second = feishu.ask('cross-channel-session', 'feishu prompt', {
signal: feishuController.signal,
onInteraction: ({ rpcId }) => received.feishu.push(rpcId),
});
await eventually(() => prompts.length === 2 && sockets.length === 2);
const activePrompt = prompts[0];
const frames = [
{
type: 'server-request', rpcId: 'cross-start', method: 'session/event',
payload: {
type: 'session/event', sessionId: 'cross-channel-session',
event: { seq: 1, type: 'turn/start', data: { turn: 1 } },
},
},
{
type: 'server-request', rpcId: 'cross-user', method: 'session/event',
payload: {
type: 'session/event', sessionId: 'cross-channel-session',
event: {
seq: 2,
type: 'user/message',
data: { source: { rpcId: activePrompt.rpcId } },
},
},
},
{
type: 'server-request', rpcId: 'cross-question', method: 'question/requested',
payload: {
type: 'question/requested', sessionId: 'cross-channel-session',
questions: [{ id: 'owner', question: 'Which channel owns this turn?' }],
},
},
];
for (const frame of frames) {
for (const { socket } of sockets) socket.frame(frame);
}
await eventually(() => received.dingtalk.length + received.feishu.length === 1);
assert.deepEqual(received, { dingtalk: ['cross-question'], feishu: [] });
dingtalkController.abort();
feishuController.abort();
await Promise.allSettled([first, second]);
});
test('a reconnect revalidates same-session ownership before accepting replayed questions', async () => {
const sockets = [];
const prompts = [];
const received = { first: [], second: [] };
let historyEvents = [];
const client = new HarnessClient({
baseUrl: 'http://127.0.0.1:3080',
workspace: '/tmp/workspace',
interactionReconnectDelayMs: 0,
createWebSocket: () => {
const socket = new FakeSocket();
sockets.push(socket);
queueMicrotask(() => socket.open());
return socket;
},
});
client.ensureRunning = async () => true;
client.rpc = async (method, payload, _timeoutMs, options) => {
if (method === 'session.history') return { events: historyEvents };
assert.equal(method, 'session.prompt');
prompts.push({ text: payload.content[0].text, rpcId: options.rpcId });
return {};
};
const firstController = new AbortController();
const secondController = new AbortController();
const first = client.ask('handoff-session', 'first', {
signal: firstController.signal,
onInteraction: ({ rpcId }) => received.first.push(rpcId),
});
await eventually(() => prompts.length === 1 && sockets.length === 1);
const second = client.ask('handoff-session', 'second', {
signal: secondController.signal,
onInteraction: ({ rpcId }) => received.second.push(rpcId),
});
await eventually(() => prompts.length === 2 && sockets.length === 2);
const firstPrompt = prompts.find(({ text }) => text === 'first');
const secondPrompt = prompts.find(({ text }) => text === 'second');
const firstTurnFrames = [
{
type: 'server-request', rpcId: 'handoff-start-one', method: 'session/event',
payload: {
type: 'session/event', sessionId: 'handoff-session',
event: { seq: 1, type: 'turn/start', data: { turn: 1 } },
},
},
{
type: 'server-request', rpcId: 'handoff-user-one', method: 'session/event',
payload: {
type: 'session/event', sessionId: 'handoff-session',
event: {
seq: 2,
type: 'user/message',
data: { source: { rpcId: firstPrompt.rpcId } },
},
},
},
{
type: 'server-request', rpcId: 'handoff-plugin-one', method: 'session/event',
payload: {
type: 'session/event', sessionId: 'handoff-session',
event: {
seq: 3,
type: 'user/message',
data: { source: { kind: 'plugin', pluginId: 'time-context' } },
},
},
},
];
for (const frame of firstTurnFrames) {
historyEvents.push({ event: frame.payload.event });
for (const socket of sockets.slice(0, 2)) socket.frame(frame);
}
const firstQuestion = {
type: 'server-request', rpcId: 'handoff-question-one', method: 'question/requested',
payload: {
type: 'question/requested', sessionId: 'handoff-session',
questions: [{ id: 'one', question: 'First turn question' }],
},
};
for (const socket of sockets.slice(0, 2)) socket.frame(firstQuestion);
await eventually(() => received.first.length === 1);
sockets[0].close(1006);
const secondTurnEvents = [
{ seq: 4, type: 'turn/end', data: { turn: 1, reason: 'completed' } },
{ seq: 5, type: 'turn/start', data: { turn: 2 } },
{
seq: 6,
type: 'user/message',
data: { source: { rpcId: secondPrompt.rpcId } },
},
];
historyEvents.push(...secondTurnEvents.map((event) => ({ event })));
for (const event of secondTurnEvents) {
sockets[1].frame({
type: 'server-request',
rpcId: `handoff-event-${event.seq}`,
method: 'session/event',
payload: { type: 'session/event', sessionId: 'handoff-session', event },
});
}
const secondQuestion = {
type: 'server-request', rpcId: 'handoff-question-two', method: 'question/requested',
payload: {
type: 'question/requested', sessionId: 'handoff-session',
questions: [{ id: 'two', question: 'Second turn question' }],
},
};
sockets[1].frame(secondQuestion);
await eventually(() => received.second.length === 1);
await eventually(() => sockets.length >= 3 && sockets[2].readyState === 1);
sockets[2].frame(secondQuestion);
await new Promise((resolve) => setTimeout(resolve, 20));
assert.deepEqual(received, {
first: ['handoff-question-one'],
second: ['handoff-question-two'],
});
firstController.abort();
secondController.abort();
await Promise.allSettled([first, second]);
});
test('a new ask adopts a replayed orphan question before its queued prompt can run', async () => {
const oldHistory = [
{ event: { seq: 1, type: 'turn/start', data: { turn: 1 } } },
{
event: {
seq: 2,
type: 'user/message',
data: { turn: 1, source: { rpcId: 'prompt-from-old-runtime' } },
},
},
];
const received = [];
const prompted = deferred();
let socket;
const controller = new AbortController();
const client = new HarnessClient({
baseUrl: 'http://127.0.0.1:3080',
workspace: '/tmp/workspace',
createWebSocket: () => {
socket = new FakeSocket();
queueMicrotask(() => {
socket.open();
socket.frame({
type: 'server-request',
rpcId: 'orphan-question',
method: 'question/requested',
payload: {
type: 'question/requested',
sessionId: 'orphan-session',
questions: [{ id: 'orphan', question: 'Recovered question' }],
},
});
});
return socket;
},
});
client.ensureRunning = async () => true;
client.rpc = async (method) => {
if (method === 'session.history') return { events: oldHistory };
assert.equal(method, 'session.prompt');
prompted.resolve();
return {};
};
const asking = client.ask('orphan-session', 'new queued prompt', {
signal: controller.signal,
onInteraction: ({ rpcId, recovered }) => received.push({ rpcId, recovered }),
});
await prompted.promise;
await eventually(() => received.length === 1);
assert.deepEqual(received, [{ rpcId: 'orphan-question', recovered: true }]);
controller.abort();
await Promise.allSettled([asking]);
});
test('an orphan approval remains fail-closed for the future approval handler', async () => {
const oldHistory = [
{ event: { seq: 1, type: 'turn/start', data: { turn: 1 } } },
{
event: {
seq: 2,
type: 'user/message',
data: { source: { rpcId: 'approval-prompt-from-old-runtime' } },
},
},
];
const received = [];
const prompted = deferred();
const controller = new AbortController();
const client = new HarnessClient({
baseUrl: 'http://127.0.0.1:3080',
workspace: '/tmp/workspace',
createWebSocket: () => {
const socket = new FakeSocket();
queueMicrotask(() => {
socket.open();
socket.frame({
type: 'server-request',
rpcId: 'orphan-approval-rpc',
method: 'approval/requested',
payload: {
type: 'approval/requested',
sessionId: 'orphan-approval-session',
approvalId: 'orphan-approval-id',
toolName: 'bash',
},
});
});
return socket;
},
});
client.ensureRunning = async () => true;
client.rpc = async (method) => {
if (method === 'session.history') return { events: oldHistory };
assert.equal(method, 'session.prompt');
prompted.resolve();
return {};
};
const asking = client.ask('orphan-approval-session', 'new queued prompt', {
signal: controller.signal,
onInteraction: (interaction) => received.push(interaction),
});
await prompted.promise;
await new Promise((resolve) => setTimeout(resolve, 20));
assert.deepEqual(received, []);
controller.abort();
await Promise.allSettled([asking]);
});

File diff suppressed because it is too large Load diff

View file

@ -2,7 +2,14 @@ import assert from 'node:assert/strict';
import test from 'node:test';
import { FeishuRuntime } from '../../../src/channels/feishu/feishu-runtime.mjs';
class FakeClient {}
class FakeClient {
static instances = [];
constructor(options) {
this.options = options;
FakeClient.instances.push(this);
}
}
class FakeDispatcher {
register(handlers) {
@ -40,24 +47,39 @@ class FakeWSClient {
function fakeLark() {
FakeWSClient.instances.length = 0;
FakeClient.instances.length = 0;
return {
Domain: { Feishu: 'feishu-domain', Lark: 'lark-domain' },
LoggerLevel: { info: 'info' },
Client: FakeClient,
EventDispatcher: FakeDispatcher,
WSClient: FakeWSClient,
defaultHttpInstance: {
request: async (options) => options,
get: async (_url, options) => options,
delete: async (_url, options) => options,
head: async (_url, options) => options,
options: async (_url, options) => options,
post: async (_url, _data, options) => options,
put: async (_url, _data, options) => options,
patch: async (_url, _data, options) => options,
},
};
}
test('FeishuRuntime becomes chat-ready only after Harness and Feishu are connected', async () => {
let harnessChecks = 0;
let harnessSignal;
const runtime = new FeishuRuntime({
lark: fakeLark(),
appId: 'cli_test',
appSecret: 'secret',
ownerOpenId: 'ou_owner',
harness: {
async ensureRunning() { harnessChecks += 1; },
async ensureRunning(options) {
harnessChecks += 1;
harnessSignal = options.signal;
},
},
state: { hasSeen: () => false },
});
@ -77,11 +99,16 @@ test('FeishuRuntime becomes chat-ready only after Harness and Feishu are connect
assert.equal(status.ready, true);
assert.equal(status.feishuLongConnectionState, 'connected');
assert.equal(status.harnessReachable, true);
assert.equal(harnessSignal.aborted, false);
assert.equal((await FakeClient.instances[0].options.httpInstance.request({
url: 'https://open.feishu.cn/test',
})).timeout, 15_000);
const stopped = await runtime.stop();
assert.equal(stopped.ready, false);
assert.equal(stopped.feishuLongConnectionState, 'idle');
assert.equal(FakeWSClient.instances[0].state, 'closed');
assert.equal(harnessSignal.aborted, true);
});
test('FeishuRuntime fails closed when the initial WebSocket handshake times out', async () => {

View file

@ -5,6 +5,160 @@ import {
HarnessReplyTracker,
} from '../../../src/channels/feishu/harness-client.mjs';
function deferred() {
let resolve;
let reject;
const promise = new Promise((promiseResolve, promiseReject) => {
resolve = promiseResolve;
reject = promiseReject;
});
return { promise, resolve, reject };
}
async function eventually(predicate, message = 'condition was not met') {
const deadline = Date.now() + 1_000;
while (Date.now() < deadline) {
if (predicate()) return;
await new Promise((resolve) => setTimeout(resolve, 5));
}
assert.fail(message);
}
class FakeSocket {
#listeners = new Map();
readyState = 0;
addEventListener(name, listener) {
const listeners = this.#listeners.get(name) ?? new Set();
listeners.add(listener);
this.#listeners.set(name, listeners);
}
removeEventListener(name, listener) {
this.#listeners.get(name)?.delete(listener);
}
open() {
if (this.readyState !== 0) return;
this.readyState = 1;
this.#emit('open', {});
}
frame(value) {
this.#emit('message', { data: JSON.stringify(value) });
}
close(code = 1000) {
if (this.readyState >= 2) return;
this.readyState = 3;
this.#emit('close', { code });
}
#emit(name, event) {
for (const listener of [...(this.#listeners.get(name) ?? [])]) listener(event);
}
}
test('interaction watcher uses the real Harness wire protocol and leaves approvals fail-closed', async () => {
const requests = [];
const opened = deferred();
let socket;
let socketUrl;
const client = new HarnessClient({
baseUrl: 'http://127.0.0.1:3080/base',
workspace: '/tmp/dsh-feishu-workspace',
fetchImpl: async (url, options) => {
requests.push({
url: url.toString(),
method: options.method,
body: JSON.parse(options.body),
});
return { ok: true, json: async () => ({ accepted: true }) };
},
createWebSocket: (url) => {
socketUrl = url;
socket = new FakeSocket();
queueMicrotask(() => socket.open());
return socket;
},
});
const controller = new AbortController();
const interactions = [];
const watching = client.watchInteractions('session-feishu', {
signal: controller.signal,
onOpen: opened.resolve,
onInteraction: (interaction) => interactions.push(interaction),
});
await opened.promise;
socket.frame({
type: 'server-request',
rpcId: 'approval-rpc',
method: 'approval/requested',
payload: {
type: 'approval/requested',
sessionId: 'session-feishu',
approvalId: 'approval-one',
toolName: 'bash',
},
});
socket.frame({
type: 'server-request',
rpcId: 'question-rpc',
method: 'question/requested',
payload: {
type: 'question/requested',
sessionId: 'session-feishu',
questions: [{
id: 'environment',
question: '请选择测试环境',
options: [{ label: '测试环境' }, { label: '生产环境' }],
}],
},
});
await eventually(() => interactions.length === 2);
assert.equal(socketUrl, 'ws://127.0.0.1:3080/api/events.mux');
assert.deepEqual(interactions.map((interaction) => ({
kind: interaction.kind,
interactionId: interaction.interactionId,
rpcId: interaction.rpcId,
sessionId: interaction.sessionId,
})), [
{
kind: 'approval',
interactionId: 'approval-one',
rpcId: 'approval-rpc',
sessionId: 'session-feishu',
},
{
kind: 'question',
interactionId: 'question-rpc',
rpcId: 'question-rpc',
sessionId: 'session-feishu',
},
]);
assert.equal(requests.length, 0, 'receiving an approval must never approve it automatically');
const result = {
ok: true,
value: {
sessionId: 'session-feishu',
answer: { answers: [{ id: 'environment', selected: ['测试环境'] }] },
},
};
assert.deepEqual(await interactions[1].respond(result), { accepted: true });
assert.deepEqual(requests, [{
url: 'http://127.0.0.1:3080/api/respond',
method: 'POST',
body: { type: 'client-response', rpcId: 'question-rpc', result },
}]);
controller.abort();
await watching;
assert.equal(socket.readyState, 3);
});
test('HarnessClient lists only absolute workspace paths', async () => {
const client = new HarnessClient({
baseUrl: 'http://127.0.0.1:3080',