feat: add model and turn control commands

This commit is contained in:
xmanrui 2026-08-20 01:28:51 +08:00
parent fc8f120e13
commit 99ce38e9a4
38 changed files with 3451 additions and 169 deletions

View file

@ -98,14 +98,24 @@ When a channel does not explicitly configure `agentPreset`, each new IM session
| Command | Description |
| --- | --- |
| `/models` | List every currently configured model, grouped by provider. |
| `/model` | Show the model used by the Session bound to this chat. |
| `/model <provider/model-id>` | Switch the model for the Session bound to this chat. |
| `/stop` | Immediately stop this chat's running task while preserving work that has not started. |
| `/steer <additional instruction>` | Inject an additional instruction into this chat's running task. |
| `/compact` | Immediately compact older context in the Session bound to the current chat. |
| `/workspace <absolute workspace path>` | Switch the current bot's Harness workspace. |
| `/workspacelist` | List workspace absolute paths that still exist on the current Harness Host. |
| `/sessionlist [workspace number or absolute path]` | List every registered session ID and title in the selected workspace; omit the argument to use the current workspace. |
| `/session <Session ID>` | Bind the current chat to an existing Harness session. |
Examples: `/compact`, `/workspace /Users/alice/projects/my-app`, `/sessionlist 2`, `/sessionlist /Users/alice/projects/my-app`, or `/session session-id`
Examples: `/models`, `/model deepseek-official/deepseek-v4-pro`, `/steer inspect only the configuration file`, `/stop`, `/compact`, `/workspace /Users/alice/projects/my-app`, `/sessionlist 2`, `/sessionlist /Users/alice/projects/my-app`, or `/session session-id`
- `/models` takes no arguments and never creates a Session. It lists every currently configured Harness model using stable, copyable `provider/model-id` values. If one provider fails, models from the remaining providers are still shown.
- Bare `/model` only displays the current Session model. `/model <provider/model-id>` accepts an exact value returned by `/models`. When the chat has no Session yet, a valid switch creates and binds a blank Session without prompting the model. The switch affects only that Session; Harness also attempts to save it as the default for future Sessions, while other existing Sessions remain unchanged.
- A model cannot be switched while a task is running or waiting for an approval or question answer. Wait for it to finish or use `/stop` first. A Session containing images cannot switch to a model that does not accept image input.
- `/stop` and `/steer` control only a running task started by this chat. Even when multiple chats bind the same Session, they do not intentionally control another chat's task. `/stop` does not delete the Session or its history, preserves queued work that has not started, and is safe to repeat.
- `/steer` accepts text only, including multiple lines. It neither creates another Session nor starts a second task. Send an ordinary message when no task is running; while an approval or question is pending, answer it first or use `/stop`.
- `/compact` acts only on the Harness Session already bound to the current chat and is never sent to the model. The bot reports the applicable status when the chat has no Session yet, the Session is generating a reply, or there is no compactable history.
- The path must be an existing absolute directory. The bot returns an actionable error and the correct usage when validation fails.
- `/workspacelist` takes no arguments. It combines the Harness global registry with the current bot's path. When that current path still exists and is safe to display, it appears first and is marked as current. Any listed path can be copied directly into `/workspace`.

View file

@ -101,14 +101,24 @@ QQ 扫码接入使用腾讯 QQBot v2 官方流程。默认腾讯授权页会把
| 命令 | 作用 |
| --- | --- |
| `/models` | 按 Provider 列出当前配置的全部可用模型。 |
| `/model` | 查看当前聊天绑定会话正在使用的模型。 |
| `/model <Provider/模型ID>` | 切换当前聊天绑定会话的模型。 |
| `/stop` | 立即停止当前聊天正在运行的任务,并保留尚未开始的排队消息。 |
| `/steer <补充指令>` | 把补充指令立即加入当前聊天正在运行的任务。 |
| `/compact` | 立即压缩当前聊天绑定会话的较早上下文。 |
| `/workspace <工作区绝对路径>` | 切换当前机器人的 Harness 工作区。 |
| `/workspacelist` | 列出当前 Harness Host 上仍然存在的工作区绝对路径。 |
| `/sessionlist [工作区序号或绝对路径]` | 列出指定工作区登记的所有会话 ID 和标题;省略参数时使用当前工作区。 |
| `/session <Session ID>` | 将当前聊天绑定到指定的已有 Harness 会话。 |
示例:`/compact`、`/workspace /Users/alice/projects/my-app`、`/sessionlist 2`、`/sessionlist /Users/alice/projects/my-app` 或 `/session session-id`
示例:`/models`、`/model deepseek-official/deepseek-v4-pro`、`/steer 只检查配置文件`、`/stop`、`/compact`、`/workspace /Users/alice/projects/my-app`、`/sessionlist 2`、`/sessionlist /Users/alice/projects/my-app` 或 `/session session-id`
- `/models` 不需要参数,也不会创建会话。它列出 Harness 当前配置的全部可用模型,使用可稳定复制的 `Provider/模型ID`;某个 Provider 查询失败时,其他 Provider 的结果仍会显示。
- `/model` 不带参数时只查看当前会话模型;带完整模型 ID 时只接受 `/models` 列出的精确值。聊天尚无会话时,有效的切换命令会创建并绑定一个空白会话,但不会触发模型回复。切换只影响当前会话;Harness 还会尝试把它保存为以后新会话的默认模型,已有其他会话不受影响。
- 正在运行任务或等待审批、问题回答时不能切换模型;请等待完成,或先使用 `/stop`。含图片的会话无法切换到不支持图片输入的模型。
- `/stop` 和 `/steer` 只控制当前聊天自己发起的运行任务,即使多个聊天绑定同一个 Session,也不会有意控制其他聊天的任务。`/stop` 不删除会话或历史,并保留尚未开始的排队消息;重复发送是安全的。
- `/steer` 只接受文字,可包含多行;它不会创建新会话或第二个任务。没有运行任务时请直接发送普通消息;等待审批或问题回答时请先处理交互,或使用 `/stop`。
- `/compact` 只作用于当前聊天已经绑定的 Harness 会话,不会把命令发送给模型。当前聊天尚未创建会话、会话正在生成回复或没有可压缩历史时,机器人会直接返回对应状态。
- 只接受已经存在的绝对目录;路径无效时机器人会返回具体提示和正确用法。
- `/workspacelist` 不需要参数。它合并 Harness 全局登记项与当前机器人的路径;当前路径仍存在且可安全显示时会排在首位并标记为“当前”。结果可直接复制到 `/workspace` 命令。

File diff suppressed because one or more lines are too long

View file

@ -16,6 +16,7 @@ import {
} from '../../../../src/channels/shared/bot-workspace-store.mjs';
import { createConnectionSupervisor } from './connection-supervisor.mjs';
import { createHarnessCommandExecutor } from '../../harness-command-executor.mjs';
import { createHarnessSessionExecutors } from '../../harness-session-coordinator.mjs';
function harnessOrigin(webServer, configured) {
if (configured !== undefined) return new URL(configured);
@ -81,6 +82,10 @@ export async function createProductionController(ctx, config = {}, internals = {
return state;
};
const commandExecutor = createHarnessCommandExecutor(ctx, internals.commandExecutor);
const { controlExecutor, sessionMaintenanceExecutor } = createHarnessSessionExecutors(ctx, {
controlExecutor: internals.controlExecutor,
sessionMaintenanceExecutor: internals.sessionMaintenanceExecutor,
});
const harness = new Harness({
baseUrl: harnessOrigin(ctx.webServer, config.harnessBaseUrl),
workspace: defaultWorkspace,
@ -88,6 +93,8 @@ export async function createProductionController(ctx, config = {}, internals = {
autostart: false,
dshBin: config.dshBin ?? 'dsh',
...(commandExecutor ? { commandExecutor } : {}),
...(controlExecutor ? { controlExecutor } : {}),
...(sessionMaintenanceExecutor ? { sessionMaintenanceExecutor } : {}),
});
const coreController = new Controller({
deviceAuth,

View file

@ -4,6 +4,7 @@ import { unlink } from 'node:fs/promises';
import * as Lark from '@larksuiteoapi/node-sdk';
import { createConnectionSupervisor } from './connection-supervisor.mjs';
import { createHarnessCommandExecutor } from '../../harness-command-executor.mjs';
import { createHarnessSessionExecutors } from '../../harness-session-coordinator.mjs';
import { verifyFeishuApp } from '../../../../src/channels/feishu/feishu-app.mjs';
import { FeishuRuntime } from '../../../../src/channels/feishu/feishu-runtime.mjs';
import { HarnessClient } from '../../../../src/channels/feishu/harness-client.mjs';
@ -107,6 +108,10 @@ export async function createProductionController(ctx, config = {}, internals = {
return stateFor(botConfig);
};
const commandExecutor = createHarnessCommandExecutor(ctx, internals.commandExecutor);
const { controlExecutor, sessionMaintenanceExecutor } = createHarnessSessionExecutors(ctx, {
controlExecutor: internals.controlExecutor,
sessionMaintenanceExecutor: internals.sessionMaintenanceExecutor,
});
const harness = new Harness({
baseUrl: harnessOrigin(ctx.webServer, config.harnessBaseUrl),
workspace: defaultWorkspace,
@ -116,6 +121,8 @@ export async function createProductionController(ctx, config = {}, internals = {
autostart: false,
dshBin: config.dshBin ?? 'dsh',
...(commandExecutor ? { commandExecutor } : {}),
...(controlExecutor ? { controlExecutor } : {}),
...(sessionMaintenanceExecutor ? { sessionMaintenanceExecutor } : {}),
});
const coreController = new Controller({

View file

@ -16,6 +16,7 @@ import {
} from '../../../../src/channels/shared/bot-workspace-store.mjs';
import { createConnectionSupervisor } from './connection-supervisor.mjs';
import { createHarnessCommandExecutor } from '../../harness-command-executor.mjs';
import { createHarnessSessionExecutors } from '../../harness-session-coordinator.mjs';
function harnessOrigin(webServer, configured) {
if (configured !== undefined) return new URL(configured);
@ -72,6 +73,10 @@ export async function createProductionController(ctx, config = {}, internals = {
return state;
};
const commandExecutor = createHarnessCommandExecutor(ctx, internals.commandExecutor);
const { controlExecutor, sessionMaintenanceExecutor } = createHarnessSessionExecutors(ctx, {
controlExecutor: internals.controlExecutor,
sessionMaintenanceExecutor: internals.sessionMaintenanceExecutor,
});
const harness = new Harness({
baseUrl: harnessOrigin(ctx.webServer, config.harnessBaseUrl),
workspace: defaultWorkspace,
@ -79,6 +84,8 @@ export async function createProductionController(ctx, config = {}, internals = {
autostart: false,
dshBin: config.dshBin ?? 'dsh',
...(commandExecutor ? { commandExecutor } : {}),
...(controlExecutor ? { controlExecutor } : {}),
...(sessionMaintenanceExecutor ? { sessionMaintenanceExecutor } : {}),
});
const coreController = new Controller({
qrAuth,

View file

@ -4,6 +4,7 @@ import { join, resolve } from 'node:path';
import { createTokenConnectionSupervisor } from './connection-supervisor.mjs';
import { createHarnessCommandExecutor } from '../../harness-command-executor.mjs';
import { createHarnessSessionExecutors } from '../../harness-session-coordinator.mjs';
import {
BotWorkspaceStore,
createBotWorkspaceScope,
@ -66,6 +67,10 @@ export async function createTokenProductionController(ctx, config, internals, de
return state;
};
const commandExecutor = createHarnessCommandExecutor(ctx, internals.commandExecutor);
const { controlExecutor, sessionMaintenanceExecutor } = createHarnessSessionExecutors(ctx, {
controlExecutor: internals.controlExecutor,
sessionMaintenanceExecutor: internals.sessionMaintenanceExecutor,
});
const harness = new ResolvedHarness({
baseUrl: harnessOrigin(ctx.webServer, config.harnessBaseUrl),
workspace: defaultWorkspace,
@ -73,6 +78,8 @@ export async function createTokenProductionController(ctx, config, internals, de
autostart: false,
dshBin: config.dshBin ?? 'dsh',
...(commandExecutor ? { commandExecutor } : {}),
...(controlExecutor ? { controlExecutor } : {}),
...(sessionMaintenanceExecutor ? { sessionMaintenanceExecutor } : {}),
});
const coreController = new ResolvedController({
credentials: ctx.credentials,

View file

@ -15,6 +15,7 @@ import {
import { createTokenConnectionSupervisor } from '../shared/connection-supervisor.mjs';
import { harnessOrigin, pluginPaths } from '../shared/production.mjs';
import { createHarnessCommandExecutor } from '../../harness-command-executor.mjs';
import { createHarnessSessionExecutors } from '../../harness-session-coordinator.mjs';
export async function createProductionController(ctx, config = {}, internals = {}) {
if (!ctx?.credentials) throw new TypeError('dsh-im slack requires ctx.credentials');
@ -51,6 +52,10 @@ export async function createProductionController(ctx, config = {}, internals = {
return state;
};
const commandExecutor = createHarnessCommandExecutor(ctx, internals.commandExecutor);
const { controlExecutor, sessionMaintenanceExecutor } = createHarnessSessionExecutors(ctx, {
controlExecutor: internals.controlExecutor,
sessionMaintenanceExecutor: internals.sessionMaintenanceExecutor,
});
const harness = new ResolvedHarness({
baseUrl: harnessOrigin(ctx.webServer, config.harnessBaseUrl),
workspace: defaultWorkspace,
@ -58,6 +63,8 @@ export async function createProductionController(ctx, config = {}, internals = {
autostart: false,
dshBin: config.dshBin ?? 'dsh',
...(commandExecutor ? { commandExecutor } : {}),
...(controlExecutor ? { controlExecutor } : {}),
...(sessionMaintenanceExecutor ? { sessionMaintenanceExecutor } : {}),
});
const coreController = new ResolvedController({
credentials: ctx.credentials,

View file

@ -16,6 +16,7 @@ import {
} from '../../../../src/channels/shared/bot-workspace-store.mjs';
import { createConnectionSupervisor } from './connection-supervisor.mjs';
import { createHarnessCommandExecutor } from '../../harness-command-executor.mjs';
import { createHarnessSessionExecutors } from '../../harness-session-coordinator.mjs';
function harnessOrigin(webServer, configured) {
if (configured !== undefined) return new URL(configured);
@ -75,6 +76,10 @@ export async function createProductionController(ctx, config = {}, internals = {
return state;
};
const commandExecutor = createHarnessCommandExecutor(ctx, internals.commandExecutor);
const { controlExecutor, sessionMaintenanceExecutor } = createHarnessSessionExecutors(ctx, {
controlExecutor: internals.controlExecutor,
sessionMaintenanceExecutor: internals.sessionMaintenanceExecutor,
});
const harness = new Harness({
baseUrl: harnessOrigin(ctx.webServer, config.harnessBaseUrl),
workspace: defaultWorkspace,
@ -82,6 +87,8 @@ export async function createProductionController(ctx, config = {}, internals = {
autostart: false,
dshBin: config.dshBin ?? 'dsh',
...(commandExecutor ? { commandExecutor } : {}),
...(controlExecutor ? { controlExecutor } : {}),
...(sessionMaintenanceExecutor ? { sessionMaintenanceExecutor } : {}),
});
const coreController = new Controller({
qrAuth,

View file

@ -16,6 +16,7 @@ import {
} from '../../../../src/channels/shared/bot-workspace-store.mjs';
import { createConnectionSupervisor } from './connection-supervisor.mjs';
import { createHarnessCommandExecutor } from '../../harness-command-executor.mjs';
import { createHarnessSessionExecutors } from '../../harness-session-coordinator.mjs';
function harnessOrigin(webServer, configured) {
if (configured !== undefined) return new URL(configured);
@ -75,6 +76,10 @@ export async function createProductionController(ctx, config = {}, internals = {
return state;
};
const commandExecutor = createHarnessCommandExecutor(ctx, internals.commandExecutor);
const { controlExecutor, sessionMaintenanceExecutor } = createHarnessSessionExecutors(ctx, {
controlExecutor: internals.controlExecutor,
sessionMaintenanceExecutor: internals.sessionMaintenanceExecutor,
});
const harness = new Harness({
baseUrl: harnessOrigin(ctx.webServer, config.harnessBaseUrl),
workspace: defaultWorkspace,
@ -82,6 +87,8 @@ export async function createProductionController(ctx, config = {}, internals = {
autostart: false,
dshBin: config.dshBin ?? 'dsh',
...(commandExecutor ? { commandExecutor } : {}),
...(controlExecutor ? { controlExecutor } : {}),
...(sessionMaintenanceExecutor ? { sessionMaintenanceExecutor } : {}),
});
const coreController = new Controller({
api,

View file

@ -16,6 +16,7 @@ import {
} from '../../../../src/channels/shared/bot-workspace-store.mjs';
import { createTokenConnectionSupervisor } from '../shared/connection-supervisor.mjs';
import { createHarnessCommandExecutor } from '../../harness-command-executor.mjs';
import { createHarnessSessionExecutors } from '../../harness-session-coordinator.mjs';
const AUTH_DIRECTORY_PATTERN = /^[a-f0-9-]{36}$/;
@ -78,6 +79,10 @@ export async function createProductionController(ctx, config = {}, internals = {
return state;
};
const commandExecutor = createHarnessCommandExecutor(ctx, internals.commandExecutor);
const { controlExecutor, sessionMaintenanceExecutor } = createHarnessSessionExecutors(ctx, {
controlExecutor: internals.controlExecutor,
sessionMaintenanceExecutor: internals.sessionMaintenanceExecutor,
});
const harness = new Harness({
baseUrl: harnessOrigin(ctx.webServer, config.harnessBaseUrl),
workspace: defaultWorkspace,
@ -85,6 +90,8 @@ export async function createProductionController(ctx, config = {}, internals = {
autostart: false,
dshBin: config.dshBin ?? 'dsh',
...(commandExecutor ? { commandExecutor } : {}),
...(controlExecutor ? { controlExecutor } : {}),
...(sessionMaintenanceExecutor ? { sessionMaintenanceExecutor } : {}),
});
const coreController = new Controller({
configStore: observedConfigStore,

View file

@ -0,0 +1,134 @@
import { randomUUID } from 'node:crypto';
function agentsFromContext(ctx) {
if (typeof ctx?.get !== 'function') return undefined;
let agents;
try {
agents = ctx.get('agents');
} catch {
return undefined;
}
if (agents === undefined || agents === null) return undefined;
if (typeof agents.get !== 'function') {
throw new TypeError('dsh-im requires a callable AgentRegistry when ctx.get("agents") is present');
}
return agents;
}
function currentOwnedTurn(agent, expectedTurn, promptRpcId) {
if (agent?.status !== 'running') return false;
const events = agent?.session?.events;
if (!Array.isArray(events)) return false;
let openTurn = null;
let owned = false;
for (const event of events) {
if (event?.type === 'turn/start') {
openTurn = event.data?.turn ?? null;
owned = false;
continue;
}
if (event?.type === 'turn/end' && event.data?.turn === openTurn) {
openTurn = null;
owned = false;
continue;
}
if (openTurn === expectedTurn
&& event?.type === 'user/message'
&& event.data?.source?.rpcId === promptRpcId) {
owned = true;
}
}
return openTurn === expectedTurn && owned;
}
function deepFreeze(value) {
if (!value || typeof value !== 'object' || Object.isFrozen(value)) return value;
for (const child of Object.values(value)) deepFreeze(child);
return Object.freeze(value);
}
function steeringMessage(text) {
return deepFreeze({
id: randomUUID(),
role: 'user',
content: [{ type: 'text', text }],
source: { kind: 'user' },
});
}
function agentBusyError(cause) {
const error = new Error('Session is busy with an active turn or maintenance task', { cause });
error.code = 'agent-busy';
return error;
}
function createControlExecutor(agents) {
return ({ sessionId, expectedTurn, promptRpcId, action, text }) => {
const agent = agents.get(sessionId);
// An unattached Session cannot be coordinated in-process. Let the client
// retain its legacy HTTP path for deployments where attachment is lazy.
if (!agent) return undefined;
if (!currentOwnedTurn(agent, expectedTurn, promptRpcId)) return false;
if (action === 'stop') {
agent.cancel({ kind: 'user' }, { keepInbox: true });
return true;
}
if (action === 'steer') {
if (typeof text !== 'string' || !text.trim()) return false;
// inject() never wakes an idle driver. Because validation and injection
// share one JS tick, this context can only target this live turn's next step.
agent.inject(steeringMessage(text));
return true;
}
throw new TypeError(`Unsupported Harness control action: ${String(action)}`);
};
}
function createSessionMaintenanceExecutor(agents) {
return ({ sessionId, operation }) => {
if (typeof operation !== 'function') throw new TypeError('maintenance operation is required');
const agent = agents.get(sessionId);
if (!agent) return operation();
try {
// runMaintenance claims the true-idle phase synchronously. A prompt or a
// second maintenance operation therefore cannot interleave with the RPC.
return agent.runMaintenance((signal) => {
if (agents.get(sessionId) !== agent) {
const error = new Error('Session agent changed before maintenance started');
error.code = 'agent-unavailable';
throw error;
}
return operation(signal);
});
} catch (error) {
throw agentBusyError(error);
}
};
}
/**
* Build optional same-process Session executors without adding a hard Cordis
* service injection. Fixtures and deployments without AgentRegistry preserve
* the existing HTTP behavior.
*/
export function createHarnessSessionExecutors(ctx, provided = {}) {
const { controlExecutor, sessionMaintenanceExecutor } = provided;
if (controlExecutor !== undefined && typeof controlExecutor !== 'function') {
throw new TypeError('controlExecutor must be a function');
}
if (sessionMaintenanceExecutor !== undefined
&& typeof sessionMaintenanceExecutor !== 'function') {
throw new TypeError('sessionMaintenanceExecutor must be a function');
}
const agents = controlExecutor && sessionMaintenanceExecutor
? undefined
: agentsFromContext(ctx);
return {
controlExecutor: controlExecutor ?? (agents ? createControlExecutor(agents) : undefined),
sessionMaintenanceExecutor: sessionMaintenanceExecutor
?? (agents ? createSessionMaintenanceExecutor(agents) : undefined),
};
}

View file

@ -10,6 +10,14 @@ import {
} from '../shared/harness-question.mjs';
import { HarnessApprovalQueue } from '../shared/harness-approval.mjs';
import { runCompactCommand } from '../shared/compact-command.mjs';
import {
isControlCommand,
runControlCommand,
} from '../shared/control-command.mjs';
import {
isModelCommand,
runModelCommand,
} from '../shared/model-command.mjs';
import { runWorkspaceCommand } from '../shared/workspace-command.mjs';
import { askInWorkspaceSession } from '../shared/workspace-session.mjs';
import {
@ -33,6 +41,10 @@ const HELP_TEXT = [
'/workspacelist 列出工作区绝对路径',
'/sessionlist [工作区序号或绝对路径] 列出会话 ID 和标题',
'/session Session ID 或当前工作区序号 将当前聊天绑定到指定会话',
'/models 列出所有可用模型',
'/model [模型ID] 查看或切换当前会话模型',
'/stop 停止当前任务',
'/steer 补充指令 纠偏当前任务',
'/status 检查连接状态',
'/help 显示本帮助',
].join('\n');
@ -246,6 +258,7 @@ export class DingtalkHarnessBridge {
#pendingInteractions = new Map();
#interactionKeys = new Map();
#interactionTasks = new Set();
#commandTasks = new Set();
#acceptedMessageIds = new Set();
#approvals;
@ -314,6 +327,37 @@ export class DingtalkHarnessBridge {
rememberConnectionTestTarget(this.#state, { sessionWebhook });
}
const pending = this.#pendingInteractions.get(key);
const promptMessage = dingtalkInboundMessage(message, {
api: this.#api,
clientId: this.#clientId,
clientSecret: this.#clientSecret,
});
const commandText = nonEmptyString(promptMessage.content) ?? '';
const commandRunner = isControlCommand(commandText)
? runControlCommand
: (isModelCommand(commandText) ? runModelCommand : null);
const addressed = String(message.conversationType) !== '2' || message?.isInAtList === true;
if (commandRunner && sessionWebhook && addressed) {
let task;
task = this.#processFastCommand(
message,
messageId,
key,
sessionWebhook,
promptMessage,
commandRunner,
).catch((error) => {
if (error?.code === 'turn-stopped' || this.#signal?.aborted) return;
this.#status.lastError = '钉钉命令处理失败。';
this.#logger.error?.('[dsh-dingtalk] failed to process a command', safeErrorDiagnostic(error));
return this.#send(sessionWebhook, CARD_ERROR_TEXT).catch(() => undefined);
}).finally(() => {
this.#acceptedMessageIds.delete(messageId);
this.#commandTasks.delete(task);
});
this.#commandTasks.add(task);
return task;
}
const approvalReply = this.#approvals.claimReply({
key,
actor: sender,
@ -412,9 +456,41 @@ export class DingtalkHarnessBridge {
pending.queue ? [pending.queue] : []
)),
...this.#interactionTasks,
...this.#commandTasks,
]);
}
async #processFastCommand(message, messageId, key, sessionWebhook, prompt, runner) {
this.#signal?.throwIfAborted();
if (this.#state.hasSeen(messageId)) return;
await this.#state.markSeen(messageId);
increment(this.#status, 'messagesReceived');
this.#status.lastMessageAt = new Date().toISOString();
const result = await runner(
nonEmptyString(prompt.content) ?? '',
this.#harness,
this.#state,
key,
{
signal: this.#signal,
hasImages: hasInboundImages(prompt),
pendingInteraction: this.#pendingInteractions.has(key)
|| this.#approvals.hasPending(key),
control: { owner: this, key },
},
);
if (result?.stopped) {
await Promise.allSettled([
this.#cancelPendingInteraction(key),
this.#approvals.closeRoute(key),
]);
}
for (const reply of result?.messages ?? [result?.message]) {
if (reply) await this.#send(sessionWebhook, reply);
}
this.#status.lastError = null;
}
async #process(message, messageId, sender, key, { alreadyRecorded = false } = {}) {
this.#signal?.throwIfAborted();
if (!alreadyRecorded) {
@ -519,6 +595,7 @@ export class DingtalkHarnessBridge {
askOptions: {
timeoutMs: this.#replyTimeoutMs,
signal: this.#signal,
control: { owner: this, key },
onUpdate: cardStarted
? (update) => cardStream.push(progressText(update))
: undefined,
@ -537,6 +614,10 @@ export class DingtalkHarnessBridge {
this.#status.lastReplyAt = new Date().toISOString();
this.#status.lastError = null;
} catch (error) {
if (error?.code === 'turn-stopped') {
if (cardStarted) await cardStream.finish('已停止。').catch(() => undefined);
return;
}
if (this.#signal?.aborted) return;
this.#status.lastError = '钉钉消息处理失败。';
this.#logger.error?.(

View file

@ -18,7 +18,15 @@ import {
} from '../shared/harness-question.mjs';
import { HarnessApprovalQueue } from '../shared/harness-approval.mjs';
import { runCompactCommand } from '../shared/compact-command.mjs';
import {
isControlCommand,
runControlCommand,
} from '../shared/control-command.mjs';
import { rememberConnectionTestTarget } from '../shared/connection-test.mjs';
import {
isModelCommand,
runModelCommand,
} from '../shared/model-command.mjs';
import { runWorkspaceCommand } from '../shared/workspace-command.mjs';
import { askInWorkspaceSession } from '../shared/workspace-session.mjs';
@ -35,6 +43,10 @@ const HELP_TEXT = [
'/workspacelist 列出工作区绝对路径',
'/sessionlist [工作区序号或绝对路径] 列出会话 ID 和标题',
'/session Session ID 或当前工作区序号 将当前聊天绑定到指定会话',
'/models 列出所有可用模型',
'/model [模型ID] 查看或切换当前会话模型',
'/stop 停止当前任务',
'/steer 补充指令 纠偏当前任务',
'/status 检查连接状态',
'/help 显示本帮助',
].join('\n');
@ -77,6 +89,7 @@ export class FeishuHarnessBridge {
#resolvedQuestionReplies = new Map();
#acceptedMessageIds = new Set();
#interactionTasks = new Set();
#commandTasks = new Set();
#approvals;
#status;
#allowedSenderOpenIds;
@ -141,6 +154,37 @@ export class FeishuHarnessBridge {
this.#acceptedMessageIds.add(messageId);
const processingReaction = this.#addReaction(messageId, 'OnIt');
const commandMessage = extractInboundMessage(event, this.#client);
const commandText = nonEmptyString(commandMessage.content) ?? '';
const commandRunner = isControlCommand(commandText)
? runControlCommand
: (isModelCommand(commandText) ? runModelCommand : null);
const addressed = event?.message?.chat_type === 'p2p'
|| (Array.isArray(event?.message?.mentions) && event.message.mentions.length > 0);
if (commandRunner && addressed) {
const processing = this.#processFastCommand(
event,
messageId,
key,
commandMessage,
commandRunner,
);
let current;
current = processing
.then(() => this.#finishReaction(messageId, processingReaction, 'DONE'))
.catch((error) => this.#handleMessageFailure(
event,
messageId,
processingReaction,
error,
))
.finally(() => {
this.#acceptedMessageIds.delete(messageId);
this.#commandTasks.delete(current);
});
this.#commandTasks.add(current);
return current;
}
if (this.#isResolvedQuestionReply(event, key)) {
const current = Promise.resolve()
.then(() => this.#discardResolvedInteractionReply(event, messageId))
@ -276,6 +320,10 @@ export class FeishuHarnessBridge {
}
async #handleMessageFailure(event, messageId, processingReaction, error) {
if (error?.code === 'turn-stopped') {
await this.#removeProcessingReaction(messageId, processingReaction);
return;
}
if (this.#signal?.aborted) {
await this.#removeProcessingReaction(messageId, processingReaction);
return;
@ -297,9 +345,41 @@ export class FeishuHarnessBridge {
pending.queue ? [pending.queue] : []
)),
...this.#interactionTasks,
...this.#commandTasks,
]);
}
async #processFastCommand(event, messageId, key, message, runner) {
this.#signal?.throwIfAborted();
if (this.#state.hasSeen(messageId)) return;
await this.#state.markSeen(messageId);
this.#status.lastMessageAt = new Date().toISOString();
this.#status.messagesReceived += 1;
const result = await runner(
nonEmptyString(message.content) ?? '',
this.#harness,
this.#state,
key,
{
signal: this.#signal,
hasImages: hasInboundImages(message),
pendingInteraction: this.#pendingInteractions.has(key)
|| this.#approvals.hasPending(key),
control: { owner: this, key },
},
);
if (result?.stopped) {
await Promise.allSettled([
this.#cancelPendingInteraction(key),
this.#approvals.closeRoute(key),
]);
}
for (const reply of result?.messages ?? [result?.message]) {
if (reply) await this.#send(event.message.chat_id, reply);
}
this.#status.lastError = null;
}
async #handle(event, key, { alreadyRecorded = false } = {}) {
this.#signal?.throwIfAborted();
const messageId = event.message.message_id;
@ -372,6 +452,7 @@ export class FeishuHarnessBridge {
return {
timeoutMs: this.#replyTimeoutMs,
signal: this.#signal,
control: { owner: this, key },
onInteraction: (interaction) => this.#handleInteraction(interaction, {
key,
actor: senderOpenId(event),

View file

@ -1,5 +1,9 @@
import { runWorkspaceCommand } from '../shared/workspace-command.mjs';
import { runCompactCommand } from '../shared/compact-command.mjs';
import {
isControlCommand,
runControlCommand,
} from '../shared/control-command.mjs';
import { rememberConnectionTestTarget } from '../shared/connection-test.mjs';
import {
harnessAnswerForQuestion,
@ -7,6 +11,10 @@ import {
validHarnessQuestion,
} from '../shared/harness-question.mjs';
import { HarnessApprovalQueue } from '../shared/harness-approval.mjs';
import {
isModelCommand,
runModelCommand,
} from '../shared/model-command.mjs';
import { askInWorkspaceSession } from '../shared/workspace-session.mjs';
import {
fetchImageBuffer,
@ -38,6 +46,10 @@ const HELP_TEXT = [
'/workspacelist 列出工作区绝对路径',
'/sessionlist [工作区序号或绝对路径] 列出会话 ID 和标题',
'/session Session ID 或当前工作区序号 将当前聊天绑定到指定会话',
'/models 列出所有可用模型',
'/model [模型ID] 查看或切换当前会话模型',
'/stop 停止当前任务',
'/steer 补充指令 纠偏当前任务',
'/status 检查连接状态',
'/help 显示本帮助',
].join('\n');
@ -134,6 +146,7 @@ export class QqHarnessBridge {
#interactionKeys = new Map();
#acceptedMessageIds = new Set();
#approvalTasks = new Set();
#commandTasks = new Set();
#approvals;
constructor({
@ -184,6 +197,34 @@ export class QqHarnessBridge {
rememberConnectionTestTarget(this.#state, message.replyTarget);
}
const pending = this.#pendingInteractions.get(key);
const commandText = safeText(message);
const commandRunner = isControlCommand(commandText)
? runControlCommand
: (isModelCommand(commandText) ? runModelCommand : null);
const allowed = this.#ownerUserOpenid === '*' || sender === this.#ownerUserOpenid;
const addressed = message.kind !== 'group'
|| message.rawEventType === 'GROUP_AT_MESSAGE_CREATE';
if (commandRunner && allowed && addressed) {
let task;
task = this.#processFastCommand(
message,
messageId,
key,
commandText,
commandRunner,
).catch((error) => {
if (error?.code === 'turn-stopped' || this.#signal?.aborted) return;
this.#status.lastError = error?.message ?? String(error);
this.#logger.error?.('[dsh-im:qq] failed to process a command:', error);
return this.#bot.sendText(message.replyTarget, '消息处理失败,请稍后重试。')
.catch(() => undefined);
}).finally(() => {
this.#acceptedMessageIds.delete(messageId);
this.#commandTasks.delete(task);
});
this.#commandTasks.add(task);
return task;
}
const approval = this.#approvals.claimReply({
key,
actor: sender,
@ -262,9 +303,35 @@ export class QqHarnessBridge {
pending.queue ? [pending.queue] : []
)),
...this.#approvalTasks,
...this.#commandTasks,
]);
}
async #processFastCommand(message, messageId, key, text, runner) {
this.#signal?.throwIfAborted();
if (this.#state.hasSeen(messageId)) return;
await this.#state.markSeen(messageId);
this.#status.messagesReceived += 1;
this.#status.lastMessageAt = new Date().toISOString();
const result = await runner(text, this.#harness, this.#state, key, {
signal: this.#signal,
hasImages: hasQqImageAttachments(message),
pendingInteraction: this.#pendingInteractions.has(key)
|| this.#approvals.hasPending(key),
control: { owner: this, key },
});
if (result?.stopped) {
await Promise.allSettled([
this.#cancelPendingInteraction(key),
this.#approvals.closeRoute(key),
]);
}
for (const reply of result?.messages ?? [result?.message]) {
if (reply) await this.#bot.sendText(message.replyTarget, reply);
}
this.#status.lastError = null;
}
async #process(message, key, { alreadyRecorded = false } = {}) {
if (this.#signal?.aborted) return;
const messageId = nonEmptyString(message?.messageId);
@ -360,6 +427,7 @@ export class QqHarnessBridge {
askOptions: {
timeoutMs: this.#replyTimeoutMs,
signal: this.#signal,
control: { owner: this, key },
onUpdate: stream ? async (update) => {
const progress = update.type === 'text'
? update.text
@ -399,6 +467,22 @@ export class QqHarnessBridge {
this.#status.lastReplyAt = new Date().toISOString();
this.#status.lastError = null;
} catch (error) {
if (error?.code === 'turn-stopped') {
if (stream) {
try {
await stream.cancel?.();
} catch (streamError) {
this.#logger.warn?.('[dsh-im:qq] unable to cancel a stopped QQ stream:', streamError);
}
try {
await this.#bot.sendText(target, '已停止。');
} catch (sendError) {
this.#logger.warn?.('[dsh-im:qq] unable to announce a stopped QQ turn:', sendError);
}
}
await this.#state.markSeen(messageId);
return;
}
stream?.cancel?.();
if (this.#signal?.aborted) return;
this.#status.lastError = error?.message ?? String(error);

View file

@ -511,7 +511,9 @@ export function createBotWorkspaceScope(harness, { botId, workspaces, state }) {
}
};
}
if ((property === 'listWorkspaces' || property === 'listWorkspaceSessions')
if ((property === 'listWorkspaces'
|| property === 'listWorkspaceSessions'
|| property === 'listModels')
&& typeof target[property] === 'function') {
return async (...args) => {
if (!isCurrentScope()) {
@ -627,6 +629,30 @@ export function createBotWorkspaceScope(harness, { botId, workspaces, state }) {
sessionGenerations.delete(sessionId);
const isCurrentSession = () => isCurrentScope()
&& generation === workspaces.generationFor(botId);
const invokeCurrentSession = async (method, args, action) => {
if (!isCurrentSession()) {
throw workspaceSessionStale(
`The bot workspace changed before this ${action} started.`,
);
}
const result = await target[method](sessionId, ...args);
if (!isCurrentSession()) {
throw workspaceSessionStale(
`The bot workspace changed while this ${action} was running.`,
);
}
return result;
};
const invokeStartedSessionMutation = async (method, args, action) => {
if (!isCurrentSession()) {
throw workspaceSessionStale(
`The bot workspace changed before this ${action} started.`,
);
}
// Once an irreversible control mutation has started, preserve its
// actual outcome even if a workspace switch commits concurrently.
return target[method](sessionId, ...args);
};
return Object.freeze({
sessionId,
async sessionExists(...args) {
@ -634,6 +660,24 @@ export function createBotWorkspaceScope(harness, { botId, workspaces, state }) {
const exists = await target.sessionExists(sessionId, ...args);
return isCurrentSession() && exists;
},
models(...args) {
return invokeCurrentSession('getSessionModels', args, 'model listing');
},
selectModel(...args) {
return invokeCurrentSession('selectSessionModel', args, 'model selection');
},
isRunning(...args) {
return invokeCurrentSession('isSessionRunning', args, 'run-state check');
},
hasActiveTurn(...args) {
return invokeCurrentSession('hasActiveTurn', args, 'turn ownership check');
},
stopActiveTurn(...args) {
return invokeStartedSessionMutation('stopActiveTurn', args, 'turn stop');
},
steerActiveTurn(...args) {
return invokeStartedSessionMutation('steerActiveTurn', args, 'turn steering');
},
ask(...args) {
if (!isCurrentSession()) {
throw workspaceSessionStale(

View file

@ -0,0 +1,84 @@
const CONTROL_COMMAND = /^\/(?:stop|steer)(?=$|\s)/iu;
const STOP_COMMAND = /^\/stop(?=$|\s)/iu;
const STOP_USAGE = '用法:/stop(不带参数)';
const STEER_USAGE = '用法:/steer <补充指令>';
const TEXT_ONLY = '控制命令仅支持纯文字,请移除图片后重试。';
function commandResult(message, extra = {}) {
return { message, ...extra };
}
function requestOptions(signal) {
return signal ? { signal } : {};
}
function boundSession(harness, state, key) {
if (typeof state?.sessionFor !== 'function') return null;
const sessionId = state.sessionFor(key);
if (typeof sessionId !== 'string' || !sessionId) return null;
if (typeof harness?.workspaceSession !== 'function') {
throw new TypeError('Harness does not support workspace sessions');
}
const session = harness.workspaceSession(sessionId);
if (!session || typeof session !== 'object') {
throw new TypeError('Harness returned an invalid workspace session');
}
return session;
}
export function isControlCommand(text) {
return typeof text === 'string' && CONTROL_COMMAND.test(text.trim());
}
export async function runControlCommand(text, harness, state, key, {
signal,
hasImages = false,
pendingInteraction = false,
control,
} = {}) {
if (!isControlCommand(text)) return null;
const command = text.trim();
const stop = STOP_COMMAND.test(command);
if (hasImages) return commandResult(TEXT_ONLY);
if (stop) {
if (!/^\/stop$/iu.test(command)) return commandResult(STOP_USAGE);
const session = boundSession(harness, state, key);
if (!session) return commandResult('当前聊天没有正在运行的任务。');
if (typeof session.stopActiveTurn !== 'function') {
throw new TypeError('Harness session does not support stopping active turns');
}
const stopped = await session.stopActiveTurn(control, requestOptions(signal));
return stopped
? commandResult('已请求停止当前任务。', { stopped: true })
: commandResult('当前聊天没有正在运行的任务。');
}
const match = /^\/steer(?:\s+([\s\S]*))?$/iu.exec(command);
const instruction = match?.[1]?.trim() ?? '';
if (!instruction) return commandResult(STEER_USAGE);
if (pendingInteraction) {
return commandResult([
'当前任务正在等待你的回答或审批。',
'',
'请先处理当前请求,或者发送 /stop 停止任务。',
].join('\n'));
}
const session = boundSession(harness, state, key);
if (!session) {
return commandResult('当前聊天没有正在运行的任务,请直接发送普通消息。');
}
if (typeof session.steerActiveTurn !== 'function') {
throw new TypeError('Harness session does not support steering active turns');
}
const steered = await session.steerActiveTurn(
instruction,
control,
requestOptions(signal),
);
return steered
? commandResult('已提交补充指令,Agent 会在下一步读取。')
: commandResult('当前聊天没有正在运行的任务,请直接发送普通消息。');
}

View file

@ -110,6 +110,10 @@ export class HarnessApprovalQueue {
this.#logger = logger;
}
hasPending(key) {
return this.#routes.get(key)?.items.some((pending) => !pending.inactive) === true;
}
claimReply({
key,
actor,

View file

@ -12,12 +12,77 @@ const interactionRegistries = new Map();
function interactionRegistry(origin) {
let registry = interactionRegistries.get(origin);
if (!registry) {
registry = { ownerships: new Map(), claims: new Map(), nextOrder: 0 };
registry = {
ownerships: new Map(),
claims: new Map(),
controls: new WeakMap(),
nextOrder: 0,
};
interactionRegistries.set(origin, registry);
}
return registry;
}
function normalizeControl(control) {
const ownerType = typeof control?.owner;
if ((ownerType !== 'object' && ownerType !== 'function')
|| control.owner === null
|| typeof control.key !== 'string'
|| !control.key) return null;
return { owner: control.owner, key: control.key };
}
function validModelSelection(value) {
return value !== null
&& typeof value === 'object'
&& typeof value.provider === 'string'
&& Boolean(value.provider)
&& typeof value.model === 'string'
&& Boolean(value.model)
&& (value.reasoningEffort === undefined || typeof value.reasoningEffort === 'string');
}
function validateModelCatalog(value, method, { session = false } = {}) {
if (!value || typeof value !== 'object'
|| !Array.isArray(value.groups)
|| !Array.isArray(value.failures)) {
throw new Error(`Harness returned an invalid response for ${method}`);
}
for (const group of value.groups) {
if (!group || typeof group !== 'object'
|| typeof group.id !== 'string' || !group.id
|| typeof group.name !== 'string' || !group.name
|| !Array.isArray(group.models)) {
throw new Error(`Harness returned an invalid response for ${method}`);
}
for (const model of group.models) {
if (!model || typeof model !== 'object'
|| typeof model.id !== 'string' || !model.id
|| typeof model.name !== 'string' || !model.name) {
throw new Error(`Harness returned an invalid response for ${method}`);
}
}
}
for (const failure of value.failures) {
if (!failure || typeof failure !== 'object'
|| typeof failure.id !== 'string' || !failure.id
|| typeof failure.name !== 'string' || !failure.name
|| (failure.message !== undefined && typeof failure.message !== 'string')) {
throw new Error(`Harness returned an invalid response for ${method}`);
}
}
if (session && (!validModelSelection(value.current) || typeof value.routable !== 'boolean')) {
throw new Error(`Harness returned an invalid response for ${method}`);
}
return value;
}
function turnStoppedError() {
const error = new Error('Harness turn was stopped before producing a text reply');
error.code = 'turn-stopped';
return error;
}
function workspacePaths(value) {
if (!Array.isArray(value?.items)) return [];
return value.items.flatMap((item) => (
@ -319,10 +384,13 @@ export class HarnessClient {
#rpcIdPrefix;
#logPrefix;
#commandExecutor;
#controlExecutor;
#sessionMaintenanceExecutor;
#managedProcess = null;
#interactionRegistry;
#interactionOwnerships;
#interactionClaims;
#controlOwnerships;
constructor({
baseUrl,
@ -336,6 +404,8 @@ export class HarnessClient {
rpcIdPrefix = 'im',
logPrefix = 'dsh-im',
commandExecutor,
controlExecutor,
sessionMaintenanceExecutor,
}) {
if (typeof createWebSocket !== 'function') {
throw new TypeError('createWebSocket must be a function');
@ -352,6 +422,13 @@ export class HarnessClient {
if (commandExecutor !== undefined && typeof commandExecutor !== 'function') {
throw new TypeError('commandExecutor must be a function');
}
if (controlExecutor !== undefined && typeof controlExecutor !== 'function') {
throw new TypeError('controlExecutor must be a function');
}
if (sessionMaintenanceExecutor !== undefined
&& typeof sessionMaintenanceExecutor !== 'function') {
throw new TypeError('sessionMaintenanceExecutor must be a function');
}
this.#baseUrl = new URL(baseUrl);
this.#workspace = workspace;
// Keep an omitted preset absent so session.create resolves the Host's current default.
@ -364,9 +441,12 @@ export class HarnessClient {
this.#rpcIdPrefix = rpcIdPrefix.trim();
this.#logPrefix = logPrefix.trim();
this.#commandExecutor = commandExecutor;
this.#controlExecutor = controlExecutor;
this.#sessionMaintenanceExecutor = sessionMaintenanceExecutor;
this.#interactionRegistry = interactionRegistry(this.#baseUrl.origin);
this.#interactionOwnerships = this.#interactionRegistry.ownerships;
this.#interactionClaims = this.#interactionRegistry.claims;
this.#controlOwnerships = this.#interactionRegistry.controls;
}
async rpc(method, payload = {}, timeoutMs = 30_000, options = {}) {
@ -443,6 +523,61 @@ export class HarnessClient {
return workspaceSessions(workspace, workspaceList.archivedSessionIds, sessionList);
}
async listModels(options = {}) {
await this.ensureRunning(options);
const value = await this.rpc('llm.models', {}, 30_000, options);
return validateModelCatalog(value, 'llm.models');
}
async getSessionModels(sessionId, options = {}) {
if (typeof sessionId !== 'string' || !sessionId) throw new TypeError('sessionId is required');
await this.ensureRunning(options);
const value = await this.rpc('session.models', { sessionId }, 30_000, options);
return validateModelCatalog(value, 'session.models', { session: true });
}
async selectSessionModel(sessionId, selection, options = {}) {
if (typeof sessionId !== 'string' || !sessionId) throw new TypeError('sessionId is required');
if (!validModelSelection(selection)) {
throw new TypeError('A provider and model are required');
}
await this.ensureRunning(options);
const operation = (maintenanceSignal) => {
const signal = maintenanceSignal && options.signal
? AbortSignal.any([maintenanceSignal, options.signal])
: (maintenanceSignal ?? options.signal);
return this.rpc('session.selectModel', {
sessionId,
provider: selection.provider,
model: selection.model,
}, 30_000, signal ? { ...options, signal } : options);
};
const value = this.#sessionMaintenanceExecutor
? await this.#sessionMaintenanceExecutor({ sessionId, operation })
: await operation();
if (!value || typeof value !== 'object' || !validModelSelection(value.selected)) {
throw new Error('Harness returned an invalid response for session.selectModel');
}
return value;
}
async isSessionRunning(sessionId, options = {}) {
if (typeof sessionId !== 'string' || !sessionId) throw new TypeError('sessionId is required');
await this.ensureRunning(options);
const value = await this.rpc('session.list', {}, 30_000, options);
if (!value || typeof value !== 'object' || !Array.isArray(value.items)) {
throw new Error('Harness returned an invalid response for session.list');
}
for (const item of value.items) {
if (!item || typeof item !== 'object'
|| typeof item.sessionId !== 'string' || !item.sessionId
|| typeof item.running !== 'boolean') {
throw new Error('Harness returned an invalid response for session.list');
}
}
return value.items.find((item) => item.sessionId === sessionId)?.running ?? false;
}
async adoptWorkspaceSession(value, options = {}) {
return adoptRegisteredWorkspaceSession(this, value, options);
}
@ -585,6 +720,154 @@ export class HarnessClient {
}
}
#registerControlOwnership(ownership) {
if (!ownership.control) return;
let routes = this.#controlOwnerships.get(ownership.control.owner);
if (!routes) {
routes = new Map();
this.#controlOwnerships.set(ownership.control.owner, routes);
}
const owners = routes.get(ownership.control.key) ?? new Set();
owners.add(ownership);
routes.set(ownership.control.key, owners);
}
#unregisterControlOwnership(ownership) {
if (!ownership.control) return;
const routes = this.#controlOwnerships.get(ownership.control.owner);
const owners = routes?.get(ownership.control.key);
owners?.delete(ownership);
if (owners?.size === 0) routes.delete(ownership.control.key);
if (routes?.size === 0) this.#controlOwnerships.delete(ownership.control.owner);
}
#controlCandidates(sessionId, control) {
const normalized = normalizeControl(control);
if (!normalized) return [];
const owners = this.#controlOwnerships.get(normalized.owner)?.get(normalized.key);
return [...(owners ?? [])]
.filter((ownership) => ownership.sessionId === sessionId)
.sort((left, right) => left.order - right.order);
}
#activeControlOwnership(sessionId, control) {
return this.#controlCandidates(sessionId, control).find((ownership) => (
ownership.started
&& ownership.active
&& !ownership.completed
&& ownership.turn !== null
)) ?? null;
}
async #refreshControlOwnership(sessionId, control, options) {
const candidates = this.#controlCandidates(sessionId, control);
// An exact local owner is mandatory before even observing the Session.
// This prevents an unrelated chat bound to the same Session from using
// run state as authority to cancel or steer somebody else's turn.
if (candidates.length === 0) return null;
try {
const history = await this.rpc(
'session.history',
{ sessionId, maxMessages: 50 },
30_000,
options,
);
this.#consumeInteractionOwnerships(sessionId, history.events ?? []);
} catch (error) {
if (!(error instanceof HarnessRpcError) || error.code !== 'session-not-found') throw error;
for (const ownership of candidates) {
ownership.active = false;
ownership.completed = true;
}
return null;
}
return this.#activeControlOwnership(sessionId, control);
}
async hasActiveTurn(sessionId, control, options = {}) {
if (typeof sessionId !== 'string' || !sessionId) throw new TypeError('sessionId is required');
return Boolean(await this.#refreshControlOwnership(sessionId, control, options));
}
async stopActiveTurn(sessionId, control, options = {}) {
if (typeof sessionId !== 'string' || !sessionId) throw new TypeError('sessionId is required');
const ownership = await this.#refreshControlOwnership(sessionId, control, options);
if (!ownership) return false;
if (ownership.stopRequested) return true;
// Re-check after the refresh await. The owning ask may have completed and
// unregistered while session.history was in flight.
if (this.#activeControlOwnership(sessionId, control) !== ownership) return false;
ownership.stopRequested = true;
try {
if (this.#controlExecutor) {
const accepted = this.#controlExecutor({
sessionId,
expectedTurn: ownership.turn,
promptRpcId: ownership.promptRpcId,
action: 'stop',
});
if (accepted && typeof accepted.then === 'function') {
throw new TypeError('controlExecutor must return synchronously');
}
if (accepted !== undefined) {
if (typeof accepted !== 'boolean') {
throw new TypeError('controlExecutor must return a boolean or undefined');
}
if (!accepted) ownership.stopRequested = false;
return accepted;
}
}
await this.rpc(
'session.cancel',
{ sessionId, keepInbox: true },
30_000,
options,
);
return true;
} catch (error) {
if (this.#activeControlOwnership(sessionId, control) === ownership) {
ownership.stopRequested = false;
}
if (error instanceof HarnessRpcError && error.code === 'session-not-found') return false;
throw error;
}
}
async steerActiveTurn(sessionId, text, control, options = {}) {
if (typeof sessionId !== 'string' || !sessionId) throw new TypeError('sessionId is required');
if (typeof text !== 'string' || !text.trim()) {
throw new TypeError('Steering text is required');
}
const ownership = await this.#refreshControlOwnership(sessionId, control, options);
if (!ownership || ownership.stopRequested) return false;
if (this.#activeControlOwnership(sessionId, control) !== ownership) return false;
if (this.#controlExecutor) {
const accepted = this.#controlExecutor({
sessionId,
expectedTurn: ownership.turn,
promptRpcId: ownership.promptRpcId,
action: 'steer',
text,
});
if (accepted && typeof accepted.then === 'function') {
throw new TypeError('controlExecutor must return synchronously');
}
if (accepted !== undefined) {
if (typeof accepted !== 'boolean') {
throw new TypeError('controlExecutor must return a boolean or undefined');
}
return accepted;
}
}
await this.rpc('session.prompt', {
sessionId,
mode: 'steer',
content: [{ type: 'text', text }],
clientTimeZone: Intl.DateTimeFormat().resolvedOptions().timeZone,
}, 30_000, options);
return true;
}
#consumeInteractionOwnerships(sessionId, entries) {
for (const ownership of this.#interactionOwnerships.get(sessionId) ?? []) {
consumeInteractionOwnership(ownership, entries);
@ -634,6 +917,7 @@ export class HarnessClient {
const onInteractionResolved = typeof options.onInteractionResolved === 'function'
? options.onInteractionResolved
: undefined;
const control = normalizeControl(options.control);
await this.ensureRunning({ signal });
const before = await this.rpc(
'session.history',
@ -655,23 +939,29 @@ export class HarnessClient {
// 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
const ownership = interactionController || control
? {
sessionId,
promptRpcId,
active: false,
started: false,
completed: false,
stopRequested: false,
turn: null,
openTurn: null,
lastSeq: baselineSeq,
reconnect: null,
order: -1,
toolCalls: new Map(),
control,
}
: null;
let interactionTask = null;
if (ownership) this.#registerInteractionOwnership(sessionId, ownership);
if (ownership) {
this.#registerInteractionOwnership(sessionId, ownership);
this.#registerControlOwnership(ownership);
}
try {
if (interactionSignal) {
@ -706,39 +996,52 @@ export class HarnessClient {
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);
try {
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;
if (ownership?.stopRequested) throw turnStoppedError();
throw new Error(
`Harness turn ended without a text reply${tracker.reason ? ` (${JSON.stringify(tracker.reason)})` : ''}`,
);
}
if (!tracker.finished) continue;
throw new Error(`Harness reply timed out after ${Math.round(timeoutMs / 1_000)} seconds`);
} catch (error) {
// Once cancellation was accepted, transport/poll failures and timeouts
// describe the convergence of that stop, not an unrelated ask failure.
if (!ownership?.stopRequested) throw error;
if (tracker.answer) return tracker.answer;
throw new Error(
`Harness turn ended without a text reply${tracker.reason ? ` (${JSON.stringify(tracker.reason)})` : ''}`,
);
if (error?.code === 'turn-stopped') throw error;
throw turnStoppedError();
}
throw new Error(`Harness reply timed out after ${Math.round(timeoutMs / 1_000)} seconds`);
} finally {
if (ownership) {
this.#unregisterControlOwnership(ownership);
this.#unregisterInteractionOwnership(sessionId, ownership);
}
interactionController?.abort(new DOMException('Harness turn finished', 'AbortError'));
if (interactionTask) await interactionTask.catch(() => undefined);
if (ownership) this.#unregisterInteractionOwnership(sessionId, ownership);
}
}
@ -775,6 +1078,10 @@ export class HarnessClient {
socket.removeEventListener('error', handleError);
signal.removeEventListener('abort', handleAbort);
if (ownership?.reconnect === close) ownership.reconnect = null;
if (signal.aborted) {
resolve();
return;
}
void callbackTail.then(() => {
const failure = error ?? callbackFailure;
if (failure) reject(failure);

View file

@ -0,0 +1,314 @@
import { splitWorkspaceCommandMessage } from './workspace-command.mjs';
import { WORKSPACE_SESSION_STALE } from './workspace-session.mjs';
import { withSessionBindingLock } from './session-binding-lock.mjs';
const MODEL_COMMAND = /^\/model(?=$|\s)/i;
const MODELS_COMMAND = /^\/models(?=$|\s)/i;
const MODEL_USAGE = '用法:/model <provider>/<model>';
const MODELS_USAGE = '用法:/models(不带参数)';
const SESSION_BINDING_CHANGED = 'session-binding-changed';
const UNSAFE_DISPLAY_TEXT_GLOBAL = /[\p{Cc}\p{Cf}\p{Zl}\p{Zp}]+/gu;
function commandResult(message) {
return {
handled: true,
message,
messages: splitWorkspaceCommandMessage(message),
};
}
function safeDisplayText(value) {
if (typeof value !== 'string') return '';
return value.replace(UNSAFE_DISPLAY_TEXT_GLOBAL, ' ').replace(/\s+/gu, ' ').trim();
}
function rpcOptions(signal) {
return signal ? { signal } : {};
}
function normalizeCatalog(value, { requireCurrent = false } = {}) {
if (!value || typeof value !== 'object'
|| !Array.isArray(value.groups) || !Array.isArray(value.failures)) {
throw new TypeError('Harness returned an invalid model catalog');
}
const groups = value.groups.map((group) => {
if (!group || typeof group !== 'object'
|| typeof group.id !== 'string' || !group.id
|| typeof group.name !== 'string' || !group.name
|| !Array.isArray(group.models)) {
throw new TypeError('Harness returned an invalid model provider group');
}
return {
id: group.id,
name: group.name,
models: group.models.map((model) => {
if (!model || typeof model !== 'object'
|| typeof model.id !== 'string' || !model.id
|| typeof model.name !== 'string' || !model.name) {
throw new TypeError('Harness returned an invalid model');
}
return { id: model.id, name: model.name };
}),
};
});
const failures = value.failures.map((failure) => {
if (!failure || typeof failure !== 'object'
|| typeof failure.id !== 'string' || !failure.id
|| typeof failure.name !== 'string' || !failure.name) {
throw new TypeError('Harness returned an invalid model provider failure');
}
return { id: failure.id, name: failure.name };
});
let current = null;
if (value.current !== undefined) {
if (!value.current || typeof value.current !== 'object'
|| typeof value.current.provider !== 'string' || !value.current.provider
|| typeof value.current.model !== 'string' || !value.current.model) {
throw new TypeError('Harness returned an invalid current model');
}
current = { provider: value.current.provider, model: value.current.model };
} else if (requireCurrent) {
throw new TypeError('Harness returned no current model');
}
return { groups, failures, current };
}
function modelId(provider, model) {
return `${provider}/${model}`;
}
function matchingModel(catalog, requested) {
for (const group of catalog.groups) {
for (const model of group.models) {
if (modelId(group.id, model.id) === requested) {
return { provider: group.id, model: model.id };
}
}
}
return null;
}
function formatCatalog(catalog) {
const currentId = catalog.current
? modelId(catalog.current.provider, catalog.current.model)
: null;
const lines = ['可用模型:'];
if (catalog.groups.length === 0) lines.push('', '当前没有可用模型。');
for (const group of catalog.groups) {
lines.push('', safeDisplayText(group.name) || safeDisplayText(group.id));
for (const model of group.models) {
const id = modelId(group.id, model.id);
lines.push(`- ${safeDisplayText(id)}${id === currentId ? '(当前)' : ''}`);
}
}
if (catalog.failures.length > 0) {
lines.push('', '以下模型提供方暂时不可用:');
for (const failure of catalog.failures) {
lines.push(`- ${safeDisplayText(failure.name) || safeDisplayText(failure.id)}`);
}
}
return lines.join('\n');
}
function currentModelMessage(current) {
return [
'当前模型:',
modelId(current.provider, current.model),
'',
'查看全部模型:/models',
'切换模型:/model <provider>/<model>',
].join('\n');
}
function noSessionMessage() {
return [
'当前聊天还没有会话。',
'',
'查看模型:/models',
'选择模型:/model <provider>/<model>',
].join('\n');
}
function errorCode(error) {
return error?.code ?? error?.failure?.code;
}
function modelErrorMessage(error, action) {
const code = errorCode(error);
if (code === 'agent-busy') {
return '当前任务正在运行,请等待完成或先发送 /stop。';
}
if (code === 'session-not-found') {
return '当前聊天绑定的会话已不存在,请重试。';
}
if (code === 'model-unavailable') {
return '无法切换到该模型。模型当前不可用,或不支持当前会话中的图片。';
}
if (code === WORKSPACE_SESSION_STALE || code === 'workspace-bot-not-found') {
return '工作区或机器人状态已发生变化,请重试。';
}
if (code === SESSION_BINDING_CHANGED) {
return '当前聊天绑定的会话已发生变化,请重试。';
}
if (code === 'cancelled' || error?.name === 'AbortError') {
return action === 'list' ? '获取模型列表已取消。' : '模型切换已取消。';
}
return action === 'list'
? '暂时无法获取模型列表,请稍后重试。'
: '模型切换失败,请稍后重试。';
}
async function boundSession(harness, state, key, options) {
if (typeof state?.sessionFor !== 'function') return null;
const sessionId = state.sessionFor(key);
if (typeof sessionId !== 'string' || !sessionId) return null;
if (typeof harness?.workspaceSession !== 'function') {
throw new TypeError('Harness does not support workspace sessions');
}
const session = harness.workspaceSession(sessionId);
if (!session || typeof session.sessionExists !== 'function') {
throw new TypeError('Harness returned an invalid workspace session');
}
if (await session.sessionExists(options)) return { sessionId, session };
if (typeof state.clearSession === 'function' && state.sessionFor(key) === sessionId) {
await state.clearSession(key);
}
return null;
}
async function sessionIsBusy(session, control, options) {
if (typeof session?.isRunning !== 'function'
|| typeof session?.hasActiveTurn !== 'function') {
throw new TypeError('Harness session does not expose run state');
}
if (await session.isRunning(options)) return true;
return Boolean(await session.hasActiveTurn(control, options));
}
async function listCatalog(harness, options) {
if (typeof harness?.listModels !== 'function') {
throw new TypeError('Harness does not support listing models');
}
return normalizeCatalog(await harness.listModels(options));
}
async function sessionCatalog(session, options) {
if (typeof session?.models !== 'function') {
throw new TypeError('Harness session does not support listing models');
}
return normalizeCatalog(await session.models(options), { requireCurrent: true });
}
function isModelsCommand(command) {
return MODELS_COMMAND.test(command);
}
export function isModelCommand(text) {
if (typeof text !== 'string') return false;
const command = text.trim();
return MODELS_COMMAND.test(command) || MODEL_COMMAND.test(command);
}
export async function runModelCommand(text, harness, state, key, options = {}) {
if (!isModelCommand(text)) return null;
const command = text.trim();
if (options.hasImages) {
return commandResult('模型命令仅支持纯文字,请移除图片后重试。');
}
const requestOptions = rpcOptions(options.signal);
if (isModelsCommand(command)) {
if (!/^\/models[ \t]*$/iu.test(command)) return commandResult(MODELS_USAGE);
try {
const bound = await boundSession(harness, state, key, requestOptions);
const catalog = bound
? await sessionCatalog(bound.session, requestOptions)
: await listCatalog(harness, requestOptions);
return commandResult(formatCatalog(catalog));
} catch (error) {
return commandResult(modelErrorMessage(error, 'list'));
}
}
const match = /^\/model(?:[ \t]+([^\s]+))?[ \t]*$/iu.exec(command);
if (!match) return commandResult(MODEL_USAGE);
const requested = match[1];
if (!requested) {
try {
const bound = await boundSession(harness, state, key, requestOptions);
if (!bound) return commandResult(noSessionMessage());
const catalog = await sessionCatalog(bound.session, requestOptions);
return commandResult(currentModelMessage(catalog.current));
} catch (error) {
return commandResult(modelErrorMessage(error, 'select'));
}
}
if (!requested.includes('/') || requested.startsWith('/') || requested.endsWith('/')) {
return commandResult(MODEL_USAGE);
}
if (options.pendingInteraction) {
return commandResult([
'当前任务正在等待你的回答或审批。',
'',
'请先处理当前请求,或者发送 /stop 停止任务。',
].join('\n'));
}
try {
return await withSessionBindingLock(state, key, async () => {
const bound = await boundSession(harness, state, key, requestOptions);
if (bound && await sessionIsBusy(bound.session, options.control, requestOptions)) {
return commandResult('当前任务正在运行,请等待完成或先发送 /stop。');
}
const catalog = bound
? await sessionCatalog(bound.session, requestOptions)
: await listCatalog(harness, requestOptions);
const selection = matchingModel(catalog, requested);
if (!selection) {
return commandResult([
`没有找到模型:${safeDisplayText(requested)}`,
'',
'请发送 /models 查看可用模型。',
].join('\n'));
}
if (bound) {
if (typeof bound.session.selectModel !== 'function') {
throw new TypeError('Harness session does not support model selection');
}
await bound.session.selectModel(selection, requestOptions);
} else {
if (typeof harness?.createSession !== 'function'
|| typeof harness?.workspaceSession !== 'function'
|| typeof state?.sessionFor !== 'function'
|| typeof state?.setSession !== 'function') {
throw new TypeError('Harness cannot create a conversation session');
}
const sessionId = await harness.createSession(requestOptions);
if (typeof sessionId !== 'string' || !sessionId) {
throw new TypeError('Harness returned an invalid session id');
}
const session = harness.workspaceSession(sessionId);
if (!session || typeof session.selectModel !== 'function') {
throw new TypeError('Harness session does not support model selection');
}
await session.selectModel(selection, requestOptions);
const currentSessionId = state.sessionFor(key);
if (typeof currentSessionId === 'string' && currentSessionId) {
const changed = new Error('Conversation binding changed during model selection');
changed.code = SESSION_BINDING_CHANGED;
throw changed;
}
if (await state.setSession(key, sessionId) === false) {
const stale = new Error('Workspace changed while binding the new session');
stale.code = WORKSPACE_SESSION_STALE;
throw stale;
}
}
return commandResult(`模型已切换为:\n${modelId(selection.provider, selection.model)}\n\n后续消息将使用该模型。`);
});
} catch (error) {
return commandResult(modelErrorMessage(error, 'select'));
}
}

View file

@ -0,0 +1,40 @@
const bindingLocks = new WeakMap();
function stateLocks(state) {
let locks = bindingLocks.get(state);
if (!locks) {
locks = new Map();
bindingLocks.set(state, locks);
}
return locks;
}
/**
* Serialize the short Session binding transaction for one conversation.
* The caller must keep long-running Session work, such as ask(), outside.
*/
export async function withSessionBindingLock(state, key, operation) {
const stateType = typeof state;
if ((stateType !== 'object' && stateType !== 'function') || state === null) {
throw new TypeError('state is required');
}
if (typeof key !== 'string' || !key) throw new TypeError('conversation key is required');
if (typeof operation !== 'function') throw new TypeError('operation is required');
const locks = stateLocks(state);
const previous = locks.get(key) ?? Promise.resolve();
let release;
const current = new Promise((resolve) => { release = resolve; });
locks.set(key, current);
await previous;
try {
return await operation();
} finally {
release();
if (locks.get(key) === current) {
locks.delete(key);
if (locks.size === 0) bindingLocks.delete(state);
}
}
}

View file

@ -1,9 +1,17 @@
import { runWorkspaceCommand } from './workspace-command.mjs';
import { runCompactCommand } from './compact-command.mjs';
import {
isControlCommand,
runControlCommand,
} from './control-command.mjs';
import {
rememberConnectionTestTarget,
sendRememberedConnectionTest,
} from './connection-test.mjs';
import {
isModelCommand,
runModelCommand,
} from './model-command.mjs';
import { askInWorkspaceSession } from './workspace-session.mjs';
import { HarnessApprovalQueue } from './harness-approval.mjs';
import {
@ -56,6 +64,7 @@ export class TextHarnessBridge {
#interactionKeys = new Map();
#acceptedMessageIds = new Set();
#approvalTasks = new Set();
#commandTasks = new Set();
#approvals;
constructor({
@ -110,6 +119,24 @@ export class TextHarnessBridge {
const key = `${kind}:${conversationId}`;
const pending = this.#pendingInteractions.get(key);
const text = cleanText(normalized.content);
const commandRunner = isControlCommand(text)
? runControlCommand
: (isModelCommand(text) ? runModelCommand : null);
if (commandRunner && (normalized.kind !== 'group' || normalized.addressed === true)) {
let task;
task = this.#processFastCommand(
normalized,
messageId,
key,
commandRunner,
).finally(() => {
this.#acceptedMessageIds.delete(messageId);
this.#commandTasks.delete(task);
});
this.#commandTasks.add(task);
return task;
}
const approval = this.#approvals.claimReply({
key,
actor: senderId,
@ -201,9 +228,48 @@ export class TextHarnessBridge {
pending.queue ? [pending.queue] : []
)),
...this.#approvalTasks,
...this.#commandTasks,
]);
}
async #processFastCommand(message, messageId, key, runner) {
if (this.#state.hasSeen(messageId)) return;
await this.#state.markSeen(messageId);
this.#status.messagesReceived += 1;
this.#status.lastMessageAt = new Date().toISOString();
const target = message.replyTarget;
try {
const result = await runner(
cleanText(message.content),
this.#harness,
this.#state,
key,
{
signal: this.#signal,
hasImages: hasInboundImages(message),
pendingInteraction: this.#pendingInteractions.has(key)
|| this.#approvals.hasPending(key),
control: { owner: this, key },
},
);
if (result?.stopped) {
await Promise.allSettled([
this.#cancelPendingInteraction(key),
this.#approvals.closeRoute(key),
]);
}
for (const reply of result?.messages ?? [result?.message]) {
if (reply) await this.#bot.sendText(target, reply);
}
this.#status.lastError = null;
} catch (error) {
if (error?.code === 'turn-stopped' || this.#signal?.aborted) return;
this.#status.lastError = error?.message ?? String(error);
this.#logger.error?.(`[dsh-im:${this.#descriptor.key}] failed to process a command:`, error);
await this.#bot.sendText(target, '消息处理失败,请稍后重试。').catch(() => undefined);
}
}
sendConnectionTest(text) {
return sendRememberedConnectionTest({
state: this.#state,
@ -250,6 +316,10 @@ export class TextHarnessBridge {
'/workspacelist 列出工作区绝对路径',
'/sessionlist [工作区序号或绝对路径] 列出会话 ID 和标题',
'/session Session ID 或当前工作区序号 将当前聊天绑定到指定会话',
'/models 列出所有可用模型',
'/model [模型ID] 查看或切换当前会话模型',
'/stop 停止当前任务',
'/steer 补充指令 纠偏当前任务',
'/status 检查连接状态',
'/help 显示本帮助',
].join('\n'));
@ -316,6 +386,7 @@ export class TextHarnessBridge {
askOptions: {
timeoutMs: this.#replyTimeoutMs,
signal: this.#signal,
control: { owner: this, key: conversationKey },
onUpdate: stream ? async (update) => {
const progress = update.type === 'text' ? update.text
: update.type === 'tool' ? `正在使用${update.name}…` : update.text;
@ -347,6 +418,16 @@ export class TextHarnessBridge {
this.#status.lastReplyAt = new Date().toISOString();
this.#status.lastError = null;
} catch (error) {
if (error?.code === 'turn-stopped') {
if (stream) {
try {
await stream.finish('已停止。');
} catch {
stream.cancel?.();
}
}
return;
}
stream?.cancel?.();
if (this.#signal?.aborted) return;
this.#status.lastError = error?.message ?? String(error);

View file

@ -1,3 +1,5 @@
import { withSessionBindingLock } from './session-binding-lock.mjs';
export const WORKSPACE_SESSION_STALE = 'workspace-session-stale';
function workspaceSession(harness, sessionId) {
@ -7,6 +9,12 @@ function workspaceSession(harness, sessionId) {
return Object.freeze({
sessionId,
sessionExists: (...args) => harness.sessionExists(sessionId, ...args),
models: (...args) => harness.getSessionModels(sessionId, ...args),
selectModel: (...args) => harness.selectSessionModel(sessionId, ...args),
isRunning: (...args) => harness.isSessionRunning(sessionId, ...args),
hasActiveTurn: (...args) => harness.hasActiveTurn(sessionId, ...args),
stopActiveTurn: (...args) => harness.stopActiveTurn(sessionId, ...args),
steerActiveTurn: (...args) => harness.steerActiveTurn(sessionId, ...args),
ask: (...args) => harness.ask(sessionId, ...args),
});
}
@ -40,16 +48,20 @@ export async function askInWorkspaceSession({
}) {
while (true) {
try {
let sessionId = state.sessionFor(key);
let session = sessionId ? workspaceSession(harness, sessionId) : null;
if (!session || !(await sessionExists(session, existsOptions))) {
sessionId = await createSession(harness, createOptions);
if (await state.setSession(key, sessionId) === false) continue;
session = workspaceSession(harness, sessionId);
}
const binding = await withSessionBindingLock(state, key, async () => {
let sessionId = state.sessionFor(key);
let session = sessionId ? workspaceSession(harness, sessionId) : null;
if (!session || !(await sessionExists(session, existsOptions))) {
sessionId = await createSession(harness, createOptions);
if (await state.setSession(key, sessionId) === false) return null;
session = workspaceSession(harness, sessionId);
}
return { sessionId, session };
});
if (!binding) continue;
return {
sessionId,
answer: await session.ask(content ?? text, askOptions),
sessionId: binding.sessionId,
answer: await binding.session.ask(content ?? text, askOptions),
};
} catch (error) {
if (error?.code !== WORKSPACE_SESSION_STALE) throw error;

View file

@ -6,6 +6,14 @@ import {
} from '../shared/harness-question.mjs';
import { HarnessApprovalQueue } from '../shared/harness-approval.mjs';
import { runCompactCommand } from '../shared/compact-command.mjs';
import {
isControlCommand,
runControlCommand,
} from '../shared/control-command.mjs';
import {
isModelCommand,
runModelCommand,
} from '../shared/model-command.mjs';
import { runWorkspaceCommand } from '../shared/workspace-command.mjs';
import { askInWorkspaceSession } from '../shared/workspace-session.mjs';
import {
@ -26,6 +34,10 @@ const HELP_TEXT = [
'/workspacelist 列出工作区绝对路径',
'/sessionlist [工作区序号或绝对路径] 列出会话 ID 和标题',
'/session Session ID 或当前工作区序号 将当前聊天绑定到指定会话',
'/models 列出所有可用模型',
'/model [模型ID] 查看或切换当前会话模型',
'/stop 停止当前任务',
'/steer 补充指令 纠偏当前任务',
'/status 检查连接状态',
'/help 显示本帮助',
].join('\n');
@ -223,6 +235,7 @@ export class WecomHarnessBridge {
#interactionKeys = new Map();
#acceptedMessageIds = new Set();
#approvalTasks = new Set();
#commandTasks = new Set();
#approvals;
#prefetchedImageCount = 0;
@ -274,6 +287,33 @@ export class WecomHarnessBridge {
rememberConnectionTestTarget(this.#state, { chatId });
}
const pending = this.#pendingInteractions.get(key);
const commandMessage = wecomInboundMessage(frame, this.#client);
const commandText = nonEmptyString(commandMessage.content) ?? '';
const commandRunner = isControlCommand(commandText)
? runControlCommand
: (isModelCommand(commandText) ? runModelCommand : null);
if (commandRunner) {
let task;
task = this.#processFastCommand(
frame,
messageId,
chatId,
key,
commandMessage,
commandRunner,
).catch((error) => {
if (error?.code === 'turn-stopped' || this.#signal?.aborted) return;
this.#status.lastError = error?.message ?? String(error);
this.#logger.error?.('[dsh-im:wecom] failed to process a command');
return this.#sendImmediate(frame, chatId, '消息处理失败,请稍后重试。')
.catch(() => undefined);
}).finally(() => {
this.#acceptedMessageIds.delete(messageId);
this.#commandTasks.delete(task);
});
this.#commandTasks.add(task);
return task;
}
const approval = this.#approvals.claimReply({
key,
actor: senderId,
@ -377,9 +417,35 @@ export class WecomHarnessBridge {
pending.queue ? [pending.queue] : []
)),
...this.#approvalTasks,
...this.#commandTasks,
]);
}
async #processFastCommand(frame, messageId, chatId, key, message, runner) {
this.#signal?.throwIfAborted();
if (this.#state.hasSeen(messageId)) return;
await this.#state.markSeen(messageId);
this.#status.messagesReceived += 1;
this.#status.lastMessageAt = new Date().toISOString();
const result = await runner(message.content, this.#harness, this.#state, key, {
signal: this.#signal,
hasImages: hasInboundImages(message),
pendingInteraction: this.#pendingInteractions.has(key)
|| this.#approvals.hasPending(key),
control: { owner: this, key },
});
if (result?.stopped) {
await Promise.allSettled([
this.#cancelPendingInteraction(key),
this.#approvals.closeRoute(key),
]);
}
for (const reply of result?.messages ?? [result?.message]) {
if (reply) await this.#sendImmediate(frame, chatId, reply);
}
this.#status.lastError = null;
}
async #sendActive(chatId, text) {
for (const chunk of splitUtf8(text)) {
this.#signal?.throwIfAborted();
@ -490,6 +556,7 @@ export class WecomHarnessBridge {
askOptions: {
timeoutMs: this.#replyTimeoutMs,
signal: this.#signal,
control: { owner: this, key },
onUpdate: streamStarted && typeof this.#client.replyStreamNonBlocking === 'function'
? async (update) => {
const progress = splitUtf8(progressText(update))[0];
@ -525,6 +592,14 @@ export class WecomHarnessBridge {
this.#status.lastReplyAt = new Date().toISOString();
this.#status.lastError = null;
} catch (error) {
if (error?.code === 'turn-stopped') {
if (streamStarted && streamId) {
await this.#client.replyStream(frame, streamId, '已停止。', true)
.catch(() => undefined);
}
await this.#state.markSeen(messageId);
return;
}
if (this.#signal?.aborted) return;
this.#status.lastError = error?.message ?? String(error);
this.#logger.error?.('[dsh-im:wecom] failed to process an inbound message');

View file

@ -11,6 +11,14 @@ import {
} from '../shared/harness-question.mjs';
import { HarnessApprovalQueue } from '../shared/harness-approval.mjs';
import { runCompactCommand } from '../shared/compact-command.mjs';
import {
isControlCommand,
runControlCommand,
} from '../shared/control-command.mjs';
import {
isModelCommand,
runModelCommand,
} from '../shared/model-command.mjs';
import { runWorkspaceCommand } from '../shared/workspace-command.mjs';
import { askInWorkspaceSession } from '../shared/workspace-session.mjs';
import {
@ -32,6 +40,10 @@ const HELP_TEXT = [
'/workspacelist 列出工作区绝对路径',
'/sessionlist [工作区序号或绝对路径] 列出会话 ID 和标题',
'/session Session ID 或当前工作区序号 将当前聊天绑定到指定会话',
'/models 列出所有可用模型',
'/model [模型ID] 查看或切换当前会话模型',
'/stop 停止当前任务',
'/steer 补充指令 纠偏当前任务',
'/status 检查连接状态',
'/help 显示本帮助',
].join('\n');
@ -85,6 +97,7 @@ export class WeixinHarnessBridge {
#interactionKeys = new Map();
#acceptedMessageIds = new Set();
#approvalTasks = new Set();
#commandTasks = new Set();
#approvals;
constructor({
@ -136,6 +149,34 @@ export class WeixinHarnessBridge {
const contextToken = nonEmptyString(message?.context_token) ?? undefined;
const runId = nonEmptyString(message?.run_id) ?? undefined;
const pending = this.#pendingInteractions.get(key);
const commandText = nonEmptyString(extractWeixinText(message)) ?? '';
const commandRunner = isControlCommand(commandText)
? runControlCommand
: (isModelCommand(commandText) ? runModelCommand : null);
if (commandRunner && sender === this.#ownerUserId) {
let task;
task = this.#processFastCommand(
message,
messageId,
key,
sender,
contextToken,
runId,
commandText,
commandRunner,
).catch((error) => {
if (error?.code === 'turn-stopped' || this.#signal?.aborted) return;
this.#status.lastError = error?.message ?? String(error);
this.#logger.error?.('[dsh-weixin] failed to process a command:', error);
return this.#send(sender, '消息处理失败,请稍后重试。', contextToken, runId)
.catch(() => undefined);
}).finally(() => {
this.#acceptedMessageIds.delete(messageId);
this.#commandTasks.delete(task);
});
this.#commandTasks.add(task);
return task;
}
const approval = this.#approvals.claimReply({
key,
actor: sender,
@ -211,9 +252,44 @@ export class WeixinHarnessBridge {
pending.queue ? [pending.queue] : []
)),
...this.#approvalTasks,
...this.#commandTasks,
]);
}
async #processFastCommand(
message,
messageId,
key,
sender,
contextToken,
runId,
text,
runner,
) {
this.#signal?.throwIfAborted();
if (this.#state.hasSeen(messageId)) return;
await this.#state.markSeen(messageId);
this.#status.messagesReceived += 1;
this.#status.lastMessageAt = new Date().toISOString();
const result = await runner(text, this.#harness, this.#state, key, {
signal: this.#signal,
hasImages: hasWeixinImageItems(message),
pendingInteraction: this.#pendingInteractions.has(key)
|| this.#approvals.hasPending(key),
control: { owner: this, key },
});
if (result?.stopped) {
await Promise.allSettled([
this.#cancelPendingInteraction(key),
this.#approvals.closeRoute(key),
]);
}
for (const reply of result?.messages ?? [result?.message]) {
if (reply) await this.#send(sender, reply, contextToken, runId);
}
this.#status.lastError = null;
}
async #process(message, key, { alreadyRecorded = false } = {}) {
this.#signal?.throwIfAborted();
const messageId = weixinMessageId(message);
@ -303,6 +379,7 @@ export class WeixinHarnessBridge {
askOptions: {
timeoutMs: this.#replyTimeoutMs,
signal: this.#signal,
control: { owner: this, key },
onInteraction: (interaction) => this.#handleInteraction(interaction, {
key,
actor: sender,
@ -324,6 +401,10 @@ export class WeixinHarnessBridge {
this.#status.lastReplyAt = new Date().toISOString();
this.#status.lastError = null;
} catch (error) {
if (error?.code === 'turn-stopped') {
await this.#state.markSeen(messageId);
return;
}
if (this.#signal?.aborted) return;
this.#status.lastError = error?.message ?? String(error);
this.#logger.error?.('[dsh-weixin] failed to process an inbound message:', error);

View file

@ -321,6 +321,43 @@ test('DingTalk executes /compact for the bound Session without prompting the mod
assert.equal(sent.at(-1).text, '暂无可压缩的历史记录。');
});
test('DingTalk lists models without prompting and help advertises all four commands', async () => {
const fixture = stateFixture();
const sent = [];
let asks = 0;
let creates = 0;
const bridge = new DingtalkHarnessBridge({
api: { sendText: async (request) => sent.push(request) },
clientId: 'ding-client',
clientSecret: 'host-secret',
harness: {
listModels: async () => ({
groups: [{
id: 'dingtalk-provider',
name: 'DingTalk Provider',
models: [{ id: 'model-one', name: 'Model One' }],
}],
failures: [],
}),
createSession: async () => { creates += 1; return 'dingtalk-session'; },
ask: async () => { asks += 1; return 'unexpected model reply'; },
},
state: fixture.state,
});
await bridge.accept(message('models-dingtalk', '/models'));
assert.match(sent.at(-1).text, /dingtalk-provider\/model-one/);
assert.equal(asks, 0);
assert.equal(creates, 0);
assert.equal(fixture.sessions.size, 0);
await bridge.accept(message('help-models-dingtalk', '/help'));
const help = sent.at(-1).text;
for (const command of ['/models', '/model', '/stop', '/steer']) {
assert.equal(help.includes(command), true, command);
}
});
test('bridge maps a DingTalk direct conversation to one persistent Harness session', async () => {
const fixture = stateFixture();
const sent = [];

View file

@ -1047,7 +1047,7 @@ test('reconnect history restores a Code Mode sub-call before replaying its appro
await Promise.allSettled([asking]);
});
test('interaction callbacks preserve frame order and watcher shutdown drains them', async () => {
test('interaction callbacks preserve frame order but watcher abort does not await a hanging callback', async () => {
const opened = deferred();
const releaseRequested = deferred();
const order = [];
@ -1102,10 +1102,10 @@ test('interaction callbacks preserve frame order and watcher shutdown drains the
let settled = false;
void watching.finally(() => { settled = true; });
await new Promise((resolve) => setTimeout(resolve, 10));
assert.equal(settled, false);
assert.equal(settled, true);
releaseRequested.resolve();
await watching;
await eventually(() => order.length === 3);
assert.deepEqual(order, ['requested:start', 'requested:end', 'resolved']);
});

View file

@ -161,6 +161,46 @@ test('Feishu executes /compact for the bound Session without prompting the model
assert.deepEqual(sent, ['暂无可压缩的历史记录。']);
});
test('Feishu lists models without prompting and help advertises all four commands', async () => {
const fixture = stateFixture();
const sent = [];
let asks = 0;
let creates = 0;
const bridge = new FeishuHarnessBridge({
client: textClient(async ({ text }) => sent.push(text)),
channel: {},
harness: {
listModels: async () => ({
groups: [{
id: 'feishu-provider',
name: 'Feishu Provider',
models: [{ id: 'model-one', name: 'Model One' }],
}],
failures: [],
}),
createSession: async () => { creates += 1; return 'feishu-session'; },
ask: async () => { asks += 1; return 'unexpected model reply'; },
},
state: fixture.state,
status: bridgeStatus(),
allowedSenderOpenIds: new Set(['ou_user']),
});
await bridge.accept(event('models-feishu', '/models'));
await bridge.waitForIdle();
assert.match(sent.at(-1), /feishu-provider\/model-one/);
assert.equal(asks, 0);
assert.equal(creates, 0);
assert.equal(fixture.sessions.size, 0);
await bridge.accept(event('help-feishu', '/help'));
await bridge.waitForIdle();
const help = sent.at(-1);
for (const command of ['/models', '/model', '/stop', '/steer']) {
assert.equal(help.includes(command), true, command);
}
});
test('bridge maps a Feishu conversation to a persistent Harness session and replies', async () => {
const sent = [];
const reactions = [];

View file

@ -259,6 +259,42 @@ test('QQ executes /compact for the bound Session without prompting the model', a
assert.equal(fixture.seen.has('compact-qq'), true);
});
test('QQ lists models without prompting and help advertises all four commands', async () => {
const fixture = stateFixture();
const sent = [];
let asks = 0;
let creates = 0;
const bridge = new QqHarnessBridge({
bot: { sendText: async (_target, text) => sent.push(text) },
ownerUserOpenid: 'owner-openid',
harness: {
listModels: async () => ({
groups: [{
id: 'qq-provider',
name: 'QQ Provider',
models: [{ id: 'model-one', name: 'Model One' }],
}],
failures: [],
}),
createSession: async () => { creates += 1; return 'qq-session'; },
ask: async () => { asks += 1; return 'unexpected model reply'; },
},
state: fixture.state,
});
await bridge.accept(message({ messageId: 'models-qq', content: '/models' }));
assert.match(sent.at(-1), /qq-provider\/model-one/);
assert.equal(asks, 0);
assert.equal(creates, 0);
assert.equal(fixture.sessions.size, 0);
await bridge.accept(message({ messageId: 'help-models-qq', content: '/help' }));
const help = sent.at(-1);
for (const command of ['/models', '/model', '/stop', '/steer']) {
assert.equal(help.includes(command), true, command);
}
});
test('QQ remembers any authorized private inbound as a connection-test target', async () => {
const fixture = stateFixture();
const sent = [];
@ -337,6 +373,80 @@ test('QQ private messages stream Harness snapshots and finalize once', async ()
assert.equal(bridge.status.messagesReplied, 1);
});
test('QQ closes an opened progress stream and announces when the Harness turn is stopped', async () => {
const fixture = stateFixture([['c2c:owner-openid', 'session-stopped']]);
const frames = [];
const sent = [];
let cancellations = 0;
let loggedErrors = 0;
const bridge = new QqHarnessBridge({
bot: {
sendText: async (_target, text) => sent.push(text),
openStream: () => ({
update: async (text) => frames.push(text),
complete: async () => frames.push('DONE'),
cancel: () => { cancellations += 1; },
}),
},
ownerUserOpenid: 'owner-openid',
harness: {
sessionExists: async () => true,
ask: async (_sessionId, _text, { onUpdate }) => {
await onUpdate({ type: 'tool', name: 'bash' });
const error = new Error('turn stopped');
error.code = 'turn-stopped';
throw error;
},
},
state: fixture.state,
logger: { warn() {}, error() { loggedErrors += 1; } },
});
await bridge.accept(message({ messageId: 'qq-stopped-stream' }));
assert.deepEqual(frames, ['正在使用bash…']);
assert.equal(cancellations, 1);
assert.deepEqual(sent, ['已停止。']);
assert.equal(loggedErrors, 0);
assert.equal(fixture.seen.has('qq-stopped-stream'), true);
});
test('QQ keeps a stopped turn terminal when stream cleanup and its notice both fail', async () => {
const fixture = stateFixture([['c2c:owner-openid', 'session-stopped-fallback']]);
let warnings = 0;
let loggedErrors = 0;
const bridge = new QqHarnessBridge({
bot: {
sendText: async () => { throw new Error('send unavailable'); },
openStream: () => ({
update: async () => {},
complete: async () => {},
cancel: () => { throw new Error('cancel unavailable'); },
}),
},
ownerUserOpenid: 'owner-openid',
harness: {
sessionExists: async () => true,
ask: async () => {
const error = new Error('turn stopped');
error.code = 'turn-stopped';
throw error;
},
},
state: fixture.state,
logger: {
warn() { warnings += 1; },
error() { loggedErrors += 1; },
},
});
await bridge.accept(message({ messageId: 'qq-stopped-stream-fallback' }));
assert.equal(warnings, 2);
assert.equal(loggedErrors, 0);
assert.equal(fixture.seen.has('qq-stopped-stream-fallback'), true);
});
test('QQ bridge accepts only the scanner and requires an at-message event in groups', async () => {
let asks = 0;
const state = {

View file

@ -0,0 +1,483 @@
import assert from 'node:assert/strict';
import test from 'node:test';
import { createBotWorkspaceScope } from '../../../src/channels/shared/bot-workspace-store.mjs';
import { HarnessClient } from '../../../src/channels/shared/harness-client.mjs';
import { WORKSPACE_SESSION_STALE } from '../../../src/channels/shared/workspace-session.mjs';
const CATALOG = {
groups: [{
id: 'deepseek-official',
name: 'DeepSeek',
models: [{ id: 'deepseek-v4', name: 'DeepSeek V4' }],
}],
failures: [],
};
function deferred() {
let resolve;
const promise = new Promise((promiseResolve) => { resolve = promiseResolve; });
return { promise, resolve };
}
function modelClient(options = {}) {
const calls = [];
const responses = new Map([
['llm.models', CATALOG],
['session.models', {
...CATALOG,
current: { provider: 'deepseek-official', model: 'deepseek-v4' },
routable: true,
}],
['session.selectModel', {
selected: { provider: 'deepseek-official', model: 'deepseek-v4' },
}],
['session.list', {
items: [
{ sessionId: 'session-one', running: true },
{ sessionId: 'session-two', running: false },
],
}],
]);
const client = new HarnessClient({
baseUrl: 'http://127.0.0.1:3981',
workspace: '/tmp/workspace',
...options,
});
client.ensureRunning = async (options) => calls.push(['ensureRunning', options]);
client.rpc = async (method, payload, timeoutMs, options) => {
calls.push(['rpc', method, payload, timeoutMs, options]);
return responses.get(method);
};
return { calls, client, responses };
}
test('HarnessClient exposes and validates model and run-state RPCs', async () => {
const { calls, client } = modelClient();
const signal = new AbortController().signal;
const options = { signal };
assert.equal(await client.listModels(options), CATALOG);
assert.deepEqual((await client.getSessionModels('session-one', options)).current, {
provider: 'deepseek-official', model: 'deepseek-v4',
});
assert.deepEqual(await client.selectSessionModel('session-one', {
provider: 'deepseek-official', model: 'deepseek-v4',
}, options), {
selected: { provider: 'deepseek-official', model: 'deepseek-v4' },
});
assert.equal(await client.isSessionRunning('session-one', options), true);
assert.equal(await client.isSessionRunning('missing', options), false);
assert.deepEqual(calls.filter(([type]) => type === 'rpc').map(([, method, payload]) => (
[method, payload]
)), [
['llm.models', {}],
['session.models', { sessionId: 'session-one' }],
['session.selectModel', {
sessionId: 'session-one',
provider: 'deepseek-official',
model: 'deepseek-v4',
}],
['session.list', {}],
['session.list', {}],
]);
assert.ok(calls.filter(([type]) => type === 'ensureRunning').every(([, value]) => value === options));
});
test('HarnessClient rejects malformed model and run-state responses', async () => {
const { client, responses } = modelClient();
responses.set('llm.models', { groups: null, failures: [] });
await assert.rejects(client.listModels(), /invalid response for llm\.models/);
responses.set('session.models', {
...CATALOG,
current: { provider: 'deepseek-official' },
routable: true,
});
await assert.rejects(
client.getSessionModels('session-one'),
/invalid response for session\.models/,
);
responses.set('session.selectModel', { selected: { provider: '', model: 'bad' } });
await assert.rejects(
client.selectSessionModel('session-one', { provider: 'p', model: 'm' }),
/invalid response for session\.selectModel/,
);
responses.set('session.list', { items: [{ sessionId: 'session-one', running: 'yes' }] });
await assert.rejects(client.isSessionRunning('session-one'), /invalid response for session\.list/);
});
test('selectSessionModel holds the maintenance executor across its RPC', async () => {
const maintenanceController = new AbortController();
const stages = [];
const { calls, client } = modelClient({
sessionMaintenanceExecutor({ sessionId, operation }) {
stages.push(['maintenance', sessionId]);
const result = operation(maintenanceController.signal);
stages.push(['operation-started']);
return result;
},
});
const callerController = new AbortController();
await client.selectSessionModel('session-one', {
provider: 'deepseek-official', model: 'deepseek-v4',
}, { signal: callerController.signal });
assert.deepEqual(stages, [
['maintenance', 'session-one'],
['operation-started'],
]);
const rpcOptions = calls.find((entry) => entry[0] === 'rpc' && entry[1] === 'session.selectModel')[4];
assert.notEqual(rpcOptions.signal, callerController.signal);
assert.equal(rpcOptions.signal.aborted, false);
maintenanceController.abort();
assert.equal(rpcOptions.signal.aborted, true, 'maintenance cancellation must abort the RPC');
});
function controlledTurn({ sessionId, initialEnd = false, controlExecutor } = {}) {
const id = sessionId ?? `session-${Math.random()}`;
const calls = [];
const admitted = deferred();
let promptRpcId = null;
let ended = initialEnd;
let answer = initialEnd ? 'already complete' : '';
let endReason = initialEnd ? 'completed' : null;
let historyFailure = null;
const client = new HarnessClient({
baseUrl: 'http://127.0.0.1:3982',
workspace: '/tmp/workspace',
...(controlExecutor ? { controlExecutor } : {}),
});
client.ensureRunning = async () => true;
const history = () => {
if (!promptRpcId) return { events: [] };
const events = [
{ event: { seq: 1, type: 'turn/start', data: { turn: 7 } } },
{ event: {
seq: 2,
type: 'user/message',
data: { turn: 7, source: { rpcId: promptRpcId } },
} },
];
if (answer) {
events.push({ event: {
seq: 3,
type: 'assistant/message',
data: {
turn: 7,
message: { content: [{ type: 'text', text: answer }] },
},
} });
}
if (ended) {
events.push({ event: {
seq: 4,
type: 'turn/end',
data: { turn: 7, reason: endReason },
} });
}
return { events };
};
client.rpc = async (method, payload, _timeoutMs, options) => {
calls.push({ method, payload, options });
if (method === 'session.history') {
if (historyFailure) throw historyFailure;
return history();
}
if (method === 'session.prompt' && payload.mode === 'queue') {
promptRpcId = options.rpcId;
admitted.resolve();
return {};
}
if (method === 'session.prompt' && payload.mode === 'steer') return {};
if (method === 'session.cancel') return {};
throw new Error(`Unexpected RPC ${method}`);
};
return {
id,
calls,
client,
admitted: admitted.promise,
setText(text) { answer = text; },
failHistory(error = new Error('history unavailable')) { historyFailure = error; },
finish({ text = '', reason = 'cancelled' } = {}) {
answer = text;
endReason = reason;
ended = true;
},
};
}
test('control methods require exact owner identity, key, and Session before any RPC', async () => {
const turn = controlledTurn();
const owner = {};
const control = { owner, key: 'direct:one' };
const asking = turn.client.ask(turn.id, 'work', { control, timeoutMs: 2_000 });
void asking.catch(() => undefined);
await turn.admitted;
for (const wrong of [
{ owner: {}, key: 'direct:one' },
{ owner, key: 'direct:two' },
null,
]) {
const before = turn.calls.length;
assert.equal(await turn.client.hasActiveTurn(turn.id, wrong), false);
assert.equal(await turn.client.stopActiveTurn(turn.id, wrong), false);
assert.equal(await turn.client.steerActiveTurn(turn.id, 'do more', wrong), false);
assert.equal(turn.calls.length, before, 'unowned controls must not observe or mutate Session');
}
const beforeOtherSession = turn.calls.length;
assert.equal(await turn.client.stopActiveTurn('different-session', control), false);
assert.equal(turn.calls.length, beforeOtherSession);
assert.equal(await turn.client.hasActiveTurn(turn.id, control), true);
assert.equal(await turn.client.stopActiveTurn(turn.id, control), true);
const cancelCount = () => turn.calls.filter(({ method }) => method === 'session.cancel').length;
assert.equal(cancelCount(), 1);
assert.equal(await turn.client.stopActiveTurn(turn.id, control), true, 'stop is idempotent');
assert.equal(cancelCount(), 1, 'an idempotent stop must not issue a second cancel RPC');
turn.finish();
await assert.rejects(asking, (error) => error?.code === 'turn-stopped');
assert.deepEqual(
turn.calls.find(({ method }) => method === 'session.cancel')?.payload,
{ sessionId: turn.id, keepInbox: true },
);
});
test('a stopped turn returns partial text instead of a generic failure', async () => {
const turn = controlledTurn();
const control = { owner: {}, key: 'direct:partial' };
const asking = turn.client.ask(turn.id, 'work', { control, timeoutMs: 2_000 });
await turn.admitted;
assert.equal(await turn.client.stopActiveTurn(turn.id, control), true);
turn.finish({ text: 'partial result' });
assert.equal(await asking, 'partial result');
});
test('accepted stop converts later polling failures to turn-stopped and preserves streamed text', async () => {
const withoutText = controlledTurn();
const firstControl = { owner: {}, key: 'direct:poll-error' };
const firstAsk = withoutText.client.ask(withoutText.id, 'work', {
control: firstControl, timeoutMs: 2_000,
});
await withoutText.admitted;
assert.equal(await withoutText.client.stopActiveTurn(withoutText.id, firstControl), true);
withoutText.failHistory();
await assert.rejects(firstAsk, (error) => error?.code === 'turn-stopped');
const withText = controlledTurn();
const secondControl = { owner: {}, key: 'direct:partial-poll-error' };
const updateSeen = deferred();
const secondAsk = withText.client.ask(withText.id, 'work', {
control: secondControl,
timeoutMs: 2_000,
onUpdate(update) {
if (update.type === 'text') updateSeen.resolve();
},
});
await withText.admitted;
withText.setText('partial before transport failure');
await updateSeen.promise;
assert.equal(await withText.client.stopActiveTurn(withText.id, secondControl), true);
withText.failHistory();
assert.equal(await secondAsk, 'partial before transport failure');
});
test('in-process control executor receives exact ownership and suppresses control RPCs', async () => {
const executions = [];
const turn = controlledTurn({
controlExecutor(request) {
executions.push(request);
return true;
},
});
const control = { owner: {}, key: 'direct:executor' };
const asking = turn.client.ask(turn.id, 'work', { control, timeoutMs: 2_000 });
await turn.admitted;
assert.equal(await turn.client.steerActiveTurn(turn.id, 'stay in this turn', control), true);
assert.equal(await turn.client.stopActiveTurn(turn.id, control), true);
assert.deepEqual(executions.map(({ sessionId, expectedTurn, action, text }) => ({
sessionId, expectedTurn, action, text,
})), [
{ sessionId: turn.id, expectedTurn: 7, action: 'steer', text: 'stay in this turn' },
{ sessionId: turn.id, expectedTurn: 7, action: 'stop', text: undefined },
]);
assert.ok(executions.every(({ promptRpcId }) => typeof promptRpcId === 'string' && promptRpcId));
assert.equal(turn.calls.some(({ method, payload }) => (
method === 'session.cancel' || (method === 'session.prompt' && payload.mode === 'steer')
)), false);
turn.finish();
await assert.rejects(asking, (error) => error?.code === 'turn-stopped');
});
test('a refused in-process control action never falls through to HTTP', async () => {
const turn = controlledTurn({ controlExecutor: () => false });
const control = { owner: {}, key: 'direct:executor-race' };
const asking = turn.client.ask(turn.id, 'work', { control, timeoutMs: 2_000 });
await turn.admitted;
assert.equal(await turn.client.steerActiveTurn(turn.id, 'must not wake', control), false);
assert.equal(await turn.client.stopActiveTurn(turn.id, control), false);
assert.equal(turn.calls.some(({ method, payload }) => (
method === 'session.cancel' || (method === 'session.prompt' && payload.mode === 'steer')
)), false);
turn.finish({ text: 'normal completion', reason: 'completed' });
assert.equal(await asking, 'normal completion');
});
test('steer uses mode steer only while the exact owned turn is still active', async () => {
const turn = controlledTurn();
const control = { owner: {}, key: 'direct:steer' };
const asking = turn.client.ask(turn.id, 'work', { control, timeoutMs: 2_000 });
await turn.admitted;
assert.equal(await turn.client.steerActiveTurn(
turn.id,
'first line\nsecond line',
control,
), true);
const steer = turn.calls.find(({ method, payload }) => (
method === 'session.prompt' && payload.mode === 'steer'
));
assert.deepEqual(steer.payload, {
sessionId: turn.id,
mode: 'steer',
content: [{ type: 'text', text: 'first line\nsecond line' }],
clientTimeZone: Intl.DateTimeFormat().resolvedOptions().timeZone,
});
turn.finish({ text: 'done', reason: 'completed' });
assert.equal(await asking, 'done');
});
test('steer refuses a turn that ended during its ownership refresh', async () => {
const turn = controlledTurn({ initialEnd: true });
const control = { owner: {}, key: 'direct:ended' };
const asking = turn.client.ask(turn.id, 'work', { control, timeoutMs: 2_000 });
await turn.admitted;
assert.equal(await turn.client.steerActiveTurn(turn.id, 'must not wake', control), false);
assert.equal(turn.calls.some(({ method, payload }) => (
method === 'session.prompt' && payload.mode === 'steer'
)), false);
assert.equal(await asking, 'already complete');
});
function scopedHarnessFixture() {
const calls = [];
let generation = 1;
let present = true;
const workspaces = {
incarnationFor() { return 'incarnation-one'; },
has() { return present; },
generationFor() { return generation; },
workspaceFor() { return '/tmp/workspace'; },
};
const harness = {
async listModels(...args) {
calls.push(['listModels', ...args]);
return CATALOG;
},
async getSessionModels(...args) {
calls.push(['getSessionModels', ...args]);
return { ...CATALOG, current: { provider: 'p', model: 'm' }, routable: true };
},
async selectSessionModel(...args) {
calls.push(['selectSessionModel', ...args]);
return { selected: args[1] };
},
async isSessionRunning(...args) {
calls.push(['isSessionRunning', ...args]);
return true;
},
async hasActiveTurn(...args) {
calls.push(['hasActiveTurn', ...args]);
return true;
},
async stopActiveTurn(...args) {
calls.push(['stopActiveTurn', ...args]);
return true;
},
async steerActiveTurn(...args) {
calls.push(['steerActiveTurn', ...args]);
return true;
},
};
const scope = createBotWorkspaceScope(harness, {
botId: 'bot-one', workspaces, state: {},
});
return {
calls,
harness,
scope,
advanceGeneration() { generation += 1; },
removeBot() { present = false; },
};
}
test('workspace session handles forward every model and control API through one generation fence', async () => {
const fixture = scopedHarnessFixture();
const session = fixture.scope.harness.workspaceSession('session-one');
const signal = new AbortController().signal;
const options = { signal };
const control = { owner: {}, key: 'direct:one' };
await session.models(options);
await session.selectModel({ provider: 'p', model: 'm' }, options);
assert.equal(await session.isRunning(options), true);
assert.equal(await session.hasActiveTurn(control, options), true);
assert.equal(await session.stopActiveTurn(control, options), true);
assert.equal(await session.steerActiveTurn('more', control, options), true);
assert.deepEqual(fixture.calls, [
['getSessionModels', 'session-one', options],
['selectSessionModel', 'session-one', { provider: 'p', model: 'm' }, options],
['isSessionRunning', 'session-one', options],
['hasActiveTurn', 'session-one', control, options],
['stopActiveTurn', 'session-one', control, options],
['steerActiveTurn', 'session-one', 'more', control, options],
]);
const methods = [
['models', [options]],
['selectModel', [{ provider: 'p', model: 'm' }, options]],
['isRunning', [options]],
['hasActiveTurn', [control, options]],
['stopActiveTurn', [control, options]],
['steerActiveTurn', ['more', control, options]],
];
for (const [method, args] of methods) {
const stale = fixture.scope.harness.workspaceSession(`stale-${method}`);
fixture.advanceGeneration();
const before = fixture.calls.length;
await assert.rejects(
stale[method](...args),
(error) => error?.code === WORKSPACE_SESSION_STALE,
method,
);
assert.equal(fixture.calls.length, before, `${method} crossed a stale generation`);
}
});
test('workspace session and listModels re-check their scope after an in-flight RPC', async () => {
const fixture = scopedHarnessFixture();
const modelGate = deferred();
fixture.harness.getSessionModels = async () => modelGate.promise;
const session = fixture.scope.harness.workspaceSession('session-race');
const listing = session.models();
fixture.advanceGeneration();
modelGate.resolve({ ...CATALOG, current: { provider: 'p', model: 'm' }, routable: true });
await assert.rejects(listing, (error) => error?.code === WORKSPACE_SESSION_STALE);
const listGate = deferred();
fixture.harness.listModels = async () => listGate.promise;
const globalListing = fixture.scope.harness.listModels();
fixture.removeBot();
listGate.resolve(CATALOG);
await assert.rejects(globalListing, (error) => error?.code === 'workspace-bot-not-found');
});

View file

@ -27,6 +27,20 @@ async function eventually(predicate, timeoutMs = 1_000) {
assert.fail('condition was not met before timeout');
}
async function within(promise, timeoutMs, messageText) {
let timer;
try {
return await Promise.race([
promise,
new Promise((_, reject) => {
timer = setTimeout(() => reject(new Error(messageText)), timeoutMs);
}),
]);
} finally {
clearTimeout(timer);
}
}
function stateFixture(initialSessions = {}) {
const sessions = new Map(Object.entries(initialSessions));
const seen = new Set();
@ -150,6 +164,125 @@ test('all four shared text channels execute /compact outside the model prompt pa
}
});
test('all four shared text channels list models locally and advertise all four commands', async () => {
for (const [name, Bridge] of [
['slack', SlackHarnessBridge],
['telegram', TelegramHarnessBridge],
['discord', DiscordHarnessBridge],
['whatsapp', WhatsappHarnessBridge],
]) {
const fixture = stateFixture();
const sent = [];
let asks = 0;
let creates = 0;
const bridge = new Bridge({
bot: { sendText: async (_target, text) => sent.push(text) },
state: fixture.state,
harness: {
listModels: async () => ({
groups: [{
id: `${name}-provider`,
name: `${name} Provider`,
models: [{ id: 'model-one', name: 'Model One' }],
}],
failures: [],
}),
createSession: async () => { creates += 1; return `${name}-session`; },
ask: async () => { asks += 1; return 'unexpected model reply'; },
},
});
await bridge.accept(message(`models-${name}`, '/models'));
assert.match(sent.at(-1), new RegExp(`${name}-provider/model-one`), name);
assert.equal(asks, 0, `${name} ask`);
assert.equal(creates, 0, `${name} create`);
assert.equal(fixture.sessions.size, 0, `${name} session binding`);
await bridge.accept(message(`help-${name}`, '/help'));
const help = sent.at(-1);
for (const command of ['/models', '/model', '/stop', '/steer']) {
assert.match(help, new RegExp(`\\${command}`), `${name} ${command}`);
}
}
});
test('/stop uses the shared command fast lane without waiting for the running prompt', async () => {
const fixture = stateFixture({ 'direct:chat-a': 'session-running' });
const askStarted = deferred();
const releaseAsk = deferred();
const sent = [];
const controls = {};
let promptSettled = false;
const session = {
sessionExists: async () => true,
ask: async (_text, options) => {
controls.prompt = options.control;
askStarted.resolve();
await releaseAsk.promise;
return '原任务完成';
},
stopActiveTurn: async (control) => {
controls.stop = control;
return true;
},
};
const bridge = createBridge({
state: fixture.state,
bot: { sendText: async (_target, text) => sent.push(text) },
harness: { workspaceSession: () => session },
});
const prompt = bridge.accept(message('running-prompt', '执行一个长任务'))
.finally(() => { promptSettled = true; });
await askStarted.promise;
await within(
bridge.accept(message('running-stop', '/stop')),
250,
'/stop waited for the ordinary conversation queue',
);
assert.equal(promptSettled, false);
assert.equal(sent.includes('已请求停止当前任务。'), true);
assert.equal(controls.stop.owner, controls.prompt.owner);
assert.equal(controls.stop.key, controls.prompt.key);
releaseAsk.resolve();
await prompt;
assert.equal(sent.at(-1), '原任务完成');
});
test('a stopped shared-channel turn closes an opened stream instead of leaving a processing placeholder', async () => {
const fixture = stateFixture({ 'direct:chat-a': 'session-running' });
const finished = [];
let cancelled = 0;
const sent = [];
const stopped = new Error('stopped');
stopped.code = 'turn-stopped';
const bridge = createBridge({
state: fixture.state,
bot: {
sendText: async (_target, text) => sent.push(text),
openStream: async () => ({
update() {},
async finish(text) { finished.push(text); },
cancel() { cancelled += 1; },
}),
},
harness: {
workspaceSession: () => ({
sessionExists: async () => true,
ask: async () => { throw stopped; },
}),
},
});
await bridge.accept(message('stopped-stream', '执行长任务'));
assert.deepEqual(finished, ['已停止。']);
assert.equal(cancelled, 0);
assert.deepEqual(sent, []);
});
test('Slack, Telegram, and Discord remember any valid direct message per bot', async () => {
for (const [name, Bridge] of [
['slack', SlackHarnessBridge],

View file

@ -153,6 +153,42 @@ test('Enterprise WeChat executes /compact for the bound Session without promptin
assert.deepEqual(transport.active, []);
});
test('Enterprise WeChat lists models without prompting and help advertises all four commands', async () => {
const store = state();
store.sessionFor = () => null;
const transport = testClient();
let asks = 0;
let creates = 0;
const bridge = new WecomHarnessBridge({
client: transport.client,
generateStreamId: () => 'models-stream',
harness: {
listModels: async () => ({
groups: [{
id: 'wecom-provider',
name: 'WeCom Provider',
models: [{ id: 'model-one', name: 'Model One' }],
}],
failures: [],
}),
createSession: async () => { creates += 1; return 'wecom-session'; },
ask: async () => { asks += 1; return 'unexpected model reply'; },
},
state: store,
});
await bridge.accept(frame({ msgid: 'models-wecom', text: { content: '/models' } }));
assert.match(transport.streamed.at(-1).content, /wecom-provider\/model-one/);
assert.equal(asks, 0);
assert.equal(creates, 0);
await bridge.accept(frame({ msgid: 'help-models-wecom', text: { content: '/help' } }));
const help = transport.streamed.at(-1).content;
for (const command of ['/models', '/model', '/stop', '/steer']) {
assert.equal(help.includes(command), true, command);
}
});
test('Enterprise WeChat messages stream Harness progress and finalize once', async () => {
const replies = [];
const active = [];

View file

@ -215,6 +215,44 @@ test('Weixin executes /compact for the bound Session without prompting the model
assert.equal(fixture.seen.has('compact-weixin'), true);
});
test('Weixin lists models without prompting and help advertises all four commands', async () => {
const fixture = stateFixture();
const sent = [];
let asks = 0;
let creates = 0;
const bridge = new WeixinHarnessBridge({
api: { sendText: async (request) => sent.push(request) },
baseUrl: 'https://ilinkai.weixin.qq.com/',
token: 'host-token',
ownerUserId: 'owner-user',
harness: {
listModels: async () => ({
groups: [{
id: 'weixin-provider',
name: 'Weixin Provider',
models: [{ id: 'model-one', name: 'Model One' }],
}],
failures: [],
}),
createSession: async () => { creates += 1; return 'weixin-session'; },
ask: async () => { asks += 1; return 'unexpected model reply'; },
},
state: fixture.state,
});
await bridge.accept(message('models-weixin', '/models'));
assert.match(sent.at(-1).text, /weixin-provider\/model-one/);
assert.equal(asks, 0);
assert.equal(creates, 0);
assert.equal(fixture.sessions.size, 0);
await bridge.accept(message('help-models-weixin', '/help'));
const help = sent.at(-1).text;
for (const command of ['/models', '/model', '/stop', '/steer']) {
assert.equal(help.includes(command), true, command);
}
});
test('bridge maps the scanning Weixin user to one persistent Harness session and echoes context_token', async () => {
const sent = [];
const asked = [];

View file

@ -0,0 +1,168 @@
import assert from 'node:assert/strict';
import test from 'node:test';
import {
isControlCommand,
runControlCommand,
} from '../src/channels/shared/control-command.mjs';
import { HarnessApprovalQueue } from '../src/channels/shared/harness-approval.mjs';
function fixture({ sessionId = 'session-one', stopped = true, steered = true } = {}) {
const calls = [];
const state = {
sessionFor(key) {
calls.push(['sessionFor', key]);
return sessionId;
},
};
const harness = {
workspaceSession(id) {
calls.push(['workspaceSession', id]);
return {
async stopActiveTurn(control, options) {
calls.push(['stopActiveTurn', id, control, options]);
return stopped;
},
async steerActiveTurn(text, control, options) {
calls.push(['steerActiveTurn', id, text, control, options]);
return steered;
},
};
},
};
return { calls, harness, state };
}
test('isControlCommand reserves valid and malformed control command forms', () => {
for (const value of [
'/stop', ' /STOP ', '/stop now', '/steer', '/StEeR do this', '/steer line one\nline two',
]) {
assert.equal(isControlCommand(value), true, value);
}
for (const value of [null, '', 'stop', '/stopping', '/steering', 'hello /stop']) {
assert.equal(isControlCommand(value), false, String(value));
}
});
test('/stop is exact, text-only, and never creates a Session', async () => {
const { calls, harness, state } = fixture();
assert.match((await runControlCommand('/stop later', harness, state, 'direct:one')).message, /用法/);
assert.match((await runControlCommand('/stop', harness, state, 'direct:one', {
hasImages: true,
})).message, /纯文字/);
assert.equal(calls.length, 0);
});
test('/stop controls only the bound active turn and marks accepted cancellation', async () => {
const active = fixture();
const signal = new AbortController().signal;
const owner = {};
const control = { owner, key: 'direct:one' };
const result = await runControlCommand(
'/STOP',
active.harness,
active.state,
'direct:one',
{ signal, pendingInteraction: true, control },
);
assert.deepEqual(result, { message: '已请求停止当前任务。', stopped: true });
assert.deepEqual(active.calls, [
['sessionFor', 'direct:one'],
['workspaceSession', 'session-one'],
['stopActiveTurn', 'session-one', control, { signal }],
]);
const inactive = fixture({ stopped: false });
const missing = await runControlCommand(
'/stop', inactive.harness, inactive.state, 'direct:one', { control },
);
assert.match(missing.message, /没有正在运行/);
assert.equal(missing.stopped, undefined);
});
test('/stop and /steer return friendly no-session messages without creating one', async () => {
const { harness, state, calls } = fixture({ sessionId: null });
assert.match(
(await runControlCommand('/stop', harness, state, 'direct:one')).message,
/没有正在运行/,
);
assert.match(
(await runControlCommand('/steer 补充', harness, state, 'direct:one')).message,
/普通消息/,
);
assert.equal(calls.some(([method]) => method === 'workspaceSession'), false);
});
test('/steer requires text, preserves multiple lines, and rejects images', async () => {
const { harness, state, calls } = fixture();
assert.match((await runControlCommand('/steer', harness, state, 'direct:one')).message, /用法/);
assert.match((await runControlCommand('/steer ', harness, state, 'direct:one')).message, /用法/);
assert.match((await runControlCommand('/steer text', harness, state, 'direct:one', {
hasImages: true,
})).message, /纯文字/);
assert.equal(calls.length, 0);
const control = { owner: {}, key: 'direct:one' };
const result = await runControlCommand(
'/steer first line\nsecond line',
harness,
state,
'direct:one',
{ control },
);
assert.match(result.message, /已提交/);
assert.deepEqual(calls.find(([method]) => method === 'steerActiveTurn'), [
'steerActiveTurn',
'session-one',
'first line\nsecond line',
control,
{},
]);
});
test('/steer never reaches Harness while an interaction is pending', async () => {
const { harness, state, calls } = fixture();
const result = await runControlCommand('/steer continue', harness, state, 'direct:one', {
pendingInteraction: true,
control: { owner: {}, key: 'direct:one' },
});
assert.match(result.message, /等待你的回答或审批/);
assert.equal(calls.length, 0);
});
test('/steer reports a lost active-turn race instead of starting new work', async () => {
const { harness, state, calls } = fixture({ steered: false });
const result = await runControlCommand('/steer continue', harness, state, 'direct:one', {
control: { owner: {}, key: 'direct:one' },
});
assert.match(result.message, /普通消息/);
assert.equal(calls.filter(([method]) => method === 'steerActiveTurn').length, 1);
});
test('HarnessApprovalQueue exposes whether a route has a live approval', async () => {
const queue = new HarnessApprovalQueue({ logger: { warn() {}, error() {} } });
const replies = [];
const interaction = {
kind: 'approval',
rpcId: 'approval-rpc',
sessionId: 'session-one',
toolCall: { callId: 'call-one', name: 'bash', arguments: '{}' },
payload: {
type: 'approval/requested',
sessionId: 'session-one',
approvalId: 'approval-one',
toolName: 'bash',
callId: 'call-one',
},
async respond() {},
};
await queue.handleRequested(interaction, {
key: 'direct:one',
actor: 'user-one',
send: async (text) => replies.push(text),
});
assert.equal(queue.hasPending('direct:one'), true);
assert.equal(queue.hasPending('direct:other'), false);
await queue.closeRoute('direct:one');
assert.equal(queue.hasPending('direct:one'), false);
});

View file

@ -0,0 +1,152 @@
import assert from 'node:assert/strict';
import test from 'node:test';
import {
createHarnessSessionExecutors,
} from '../plugin-src/host/harness-session-coordinator.mjs';
function contextWith(registry) {
return { get: (name) => name === 'agents' ? registry : undefined };
}
function liveAgent(overrides = {}) {
const calls = [];
const agent = {
status: 'running',
session: {
events: [
{ seq: 1, type: 'turn/start', data: { turn: 6 } },
{ seq: 2, type: 'user/message', data: { source: { rpcId: 'older' } } },
{ seq: 3, type: 'turn/end', data: { turn: 6, reason: 'completed' } },
{ seq: 4, type: 'turn/start', data: { turn: 7 } },
{ seq: 5, type: 'user/message', data: { source: { rpcId: 'prompt-owned' } } },
],
},
cancel(...args) { calls.push(['cancel', ...args]); },
inject(...args) { calls.push(['inject', ...args]); },
steer() { throw new Error('agent.steer must never be called'); },
...overrides,
};
return { agent, calls };
}
test('Host control executor validates and mutates the exact open owned turn synchronously', () => {
const { agent, calls } = liveAgent();
const registry = { get: (sessionId) => sessionId === 'session-one' ? agent : undefined };
const { controlExecutor } = createHarnessSessionExecutors(contextWith(registry));
assert.equal(controlExecutor({
sessionId: 'session-one',
expectedTurn: 7,
promptRpcId: 'prompt-owned',
action: 'steer',
text: '先检查日志\n再继续',
}), true);
assert.equal(calls.length, 1, 'inject must finish before the executor returns');
const message = calls[0][1];
assert.equal(calls[0][0], 'inject');
assert.equal(message.role, 'user');
assert.deepEqual(message.content, [{ type: 'text', text: '先检查日志\n再继续' }]);
assert.deepEqual(message.source, { kind: 'user' });
assert.match(message.id, /^[0-9a-f-]{36}$/i);
assert.equal(Object.isFrozen(message), true);
assert.equal(Object.isFrozen(message.content), true);
assert.equal(Object.isFrozen(message.content[0]), true);
assert.equal(controlExecutor({
sessionId: 'session-one',
expectedTurn: 7,
promptRpcId: 'prompt-owned',
action: 'stop',
}), true);
assert.deepEqual(calls[1], ['cancel', { kind: 'user' }, { keepInbox: true }]);
});
test('Host control executor refuses idle, replaced, closed, and foreign turns without waking', () => {
const { agent, calls } = liveAgent();
const { controlExecutor } = createHarnessSessionExecutors(contextWith({ get: () => agent }));
const request = {
sessionId: 'session-one',
expectedTurn: 7,
promptRpcId: 'prompt-owned',
action: 'steer',
text: 'must not wake',
};
assert.equal(controlExecutor({ ...request, promptRpcId: 'foreign' }), false);
assert.equal(controlExecutor({ ...request, expectedTurn: 6 }), false);
agent.status = 'idle';
assert.equal(controlExecutor(request), false);
agent.status = 'running';
agent.session.events.push(
{ seq: 6, type: 'turn/end', data: { turn: 7, reason: 'cancelled' } },
{ seq: 7, type: 'turn/start', data: { turn: 8 } },
{ seq: 8, type: 'user/message', data: { source: { rpcId: 'someone-else' } } },
);
assert.equal(controlExecutor(request), false);
assert.deepEqual(calls, []);
});
test('Host executors preserve HTTP fallback when AgentRegistry or attachment is absent', async () => {
assert.deepEqual(createHarnessSessionExecutors({}), {
controlExecutor: undefined,
sessionMaintenanceExecutor: undefined,
});
assert.deepEqual(createHarnessSessionExecutors({ get() { throw new Error('not injected'); } }), {
controlExecutor: undefined,
sessionMaintenanceExecutor: undefined,
});
const { controlExecutor, sessionMaintenanceExecutor } = createHarnessSessionExecutors(
contextWith({ get: () => undefined }),
);
assert.equal(controlExecutor({
sessionId: 'cold-session', expectedTurn: 1, promptRpcId: 'rpc', action: 'stop',
}), undefined);
let operated = false;
assert.equal(await sessionMaintenanceExecutor({
sessionId: 'cold-session',
operation: async (signal) => {
operated = true;
assert.equal(signal, undefined);
return 'fallback';
},
}), 'fallback');
assert.equal(operated, true);
});
test('Host maintenance executor claims idle synchronously and reports busy with a stable code', async () => {
const maintenanceController = new AbortController();
let operationStarted = false;
const agent = {
runMaintenance(operation) {
operationStarted = true;
return operation(maintenanceController.signal);
},
};
const { sessionMaintenanceExecutor } = createHarnessSessionExecutors(
contextWith({ get: () => agent }),
);
const value = sessionMaintenanceExecutor({
sessionId: 'session-one',
operation: async (signal) => {
assert.equal(operationStarted, true);
assert.equal(signal, maintenanceController.signal);
return 'selected';
},
});
assert.equal(operationStarted, true, 'maintenance must be claimed in the calling tick');
assert.equal(await value, 'selected');
const busy = new Error('already active');
agent.runMaintenance = () => { throw busy; };
assert.throws(() => sessionMaintenanceExecutor({
sessionId: 'session-one', operation: async () => undefined,
}), (error) => error?.code === 'agent-busy' && error.cause === busy);
const rpcFailure = new Error('model unavailable');
agent.runMaintenance = (operation) => operation(maintenanceController.signal);
await assert.rejects(sessionMaintenanceExecutor({
sessionId: 'session-one', operation: async () => { throw rpcFailure; },
}), (error) => error === rpcFailure, 'async RPC failures must not be mislabeled as busy');
});

413
test/model-command.test.mjs Normal file
View file

@ -0,0 +1,413 @@
import assert from 'node:assert/strict';
import test from 'node:test';
import {
isModelCommand,
runModelCommand,
} from '../src/channels/shared/model-command.mjs';
const CATALOG = Object.freeze({
groups: [
{
id: 'deepseek-official',
name: 'DeepSeek',
models: [
{ id: 'deepseek-v4-flash', name: 'DeepSeek V4 Flash' },
{ id: 'deepseek-v4-pro', name: 'DeepSeek V4 Pro' },
],
},
{
id: 'openrouter',
name: 'OpenRouter',
models: [{ id: 'anthropic/claude-sonnet-4', name: 'Claude Sonnet 4' }],
},
],
failures: [],
});
function deferred() {
let resolve;
const promise = new Promise((settle) => { resolve = settle; });
return { promise, resolve };
}
function fixture({
initialSessionId = null,
existing = true,
globalCatalog = CATALOG,
sessionCatalog = {
...CATALOG,
current: { provider: 'deepseek-official', model: 'deepseek-v4-flash' },
routable: true,
},
running = false,
activeTurn = false,
selectionError = null,
selectModelHook,
setSessionResult,
} = {}) {
const calls = [];
let boundId = initialSessionId;
const session = (sessionId) => ({
async sessionExists(options) {
calls.push(['sessionExists', sessionId, options]);
return typeof existing === 'function' ? existing(sessionId) : existing;
},
async models(options) {
calls.push(['models', sessionId, options]);
if (sessionCatalog instanceof Error) throw sessionCatalog;
return sessionCatalog;
},
async isRunning(options) {
calls.push(['isRunning', sessionId, options]);
return running;
},
async hasActiveTurn(control, options) {
calls.push(['hasActiveTurn', sessionId, control, options]);
return activeTurn;
},
async selectModel(selection, options) {
calls.push(['selectModel', sessionId, selection, options]);
if (selectModelHook) await selectModelHook({ sessionId, selection, options });
if (selectionError) throw selectionError;
return { selected: selection };
},
});
const state = {
sessionFor(key) {
calls.push(['sessionFor', key]);
return boundId;
},
async setSession(key, sessionId) {
calls.push(['setSession', key, sessionId]);
if (setSessionResult === false) return false;
boundId = sessionId;
return setSessionResult;
},
async clearSession(key) {
calls.push(['clearSession', key]);
boundId = null;
},
};
const harness = {
async listModels(options) {
calls.push(['listModels', options]);
if (globalCatalog instanceof Error) throw globalCatalog;
return globalCatalog;
},
workspaceSession(sessionId) {
calls.push(['workspaceSession', sessionId]);
return session(sessionId);
},
async createSession(options) {
calls.push(['createSession', options]);
return 'session-created';
},
};
return { calls, harness, state, boundId: () => boundId };
}
test('isModelCommand recognizes only /models and /model command prefixes', () => {
for (const command of [
'/models', ' /MODELS ', '/models ignored', '/model', '/MoDeL openai/gpt-5',
]) {
assert.equal(isModelCommand(command), true, command);
}
for (const value of [null, '', 'model', '/modelx', '/modelsx', 'hello /models']) {
assert.equal(isModelCommand(value), false, String(value));
}
});
test('/models lists the global catalog without creating a Session', async () => {
const { calls, harness, state } = fixture();
const signal = new AbortController().signal;
const result = await runModelCommand('/MODELS', harness, state, 'direct:one', { signal });
assert.match(result.message, /DeepSeek/);
assert.match(result.message, /deepseek-official\/deepseek-v4-flash/);
assert.match(result.message, /openrouter\/anthropic\/claude-sonnet-4/);
assert.deepEqual(calls, [
['sessionFor', 'direct:one'],
['listModels', { signal }],
]);
});
test('/models marks the current Session model and contains provider-local failures', async () => {
const sessionCatalog = {
groups: CATALOG.groups,
current: { provider: 'deepseek-official', model: 'deepseek-v4-pro' },
routable: true,
failures: [{
id: 'private-provider',
name: 'Private Provider',
message: 'https://private.example.invalid failed with secret=abc',
}],
};
const { harness, state } = fixture({ initialSessionId: 'session-one', sessionCatalog });
const result = await runModelCommand('/models', harness, state, 'direct:one');
assert.match(result.message, /deepseek-official\/deepseek-v4-pro(当前)/);
assert.match(result.message, /Private Provider/);
assert.doesNotMatch(result.message, /private\.example|secret=abc/);
});
test('/models validates its no-argument and text-only syntax', async () => {
const { harness, state } = fixture();
assert.match(
(await runModelCommand('/models openai', harness, state, 'direct:one')).message,
/不带参数/,
);
assert.match(
(await runModelCommand('/models', harness, state, 'direct:one', { hasImages: true })).message,
/仅支持纯文字/,
);
});
test('/models splits a long catalog into lossless 1,800-character messages', async () => {
const longCatalog = {
groups: [{
id: 'provider',
name: 'Large Provider',
models: Array.from({ length: 80 }, (_, index) => ({
id: `model-${String(index).padStart(3, '0')}-${'x'.repeat(40)}`,
name: `Model ${index}`,
})),
}],
failures: [],
};
const { harness, state } = fixture({ globalCatalog: longCatalog });
const result = await runModelCommand('/models', harness, state, 'direct:one');
assert.ok(result.messages.length > 1);
assert.ok(result.messages.every((message) => message.length <= 1_800));
assert.equal(result.messages.join(''), result.message);
assert.match(result.message, /provider\/model-079/);
});
test('/model reports current state without creating or selecting', async () => {
const missing = fixture();
const noSession = await runModelCommand('/model', missing.harness, missing.state, 'direct:one');
assert.match(noSession.message, /还没有会话/);
assert.equal(missing.calls.some(([name]) => name === 'createSession'), false);
const existingFixture = fixture({ initialSessionId: 'session-one' });
const existing = await runModelCommand(
'/MODEL',
existingFixture.harness,
existingFixture.state,
'direct:one',
);
assert.match(existing.message, /deepseek-official\/deepseek-v4-flash/);
assert.equal(existingFixture.calls.some(([name]) => name === 'selectModel'), false);
});
test('/model uses an exact catalog ID and preserves slashes inside the model ID', async () => {
const { calls, harness, state } = fixture({ initialSessionId: 'session-one' });
const control = Object.freeze({ route: 'direct:one' });
const result = await runModelCommand(
'/model openrouter/anthropic/claude-sonnet-4',
harness,
state,
'direct:one',
{ control },
);
assert.match(result.message, /openrouter\/anthropic\/claude-sonnet-4/);
assert.deepEqual(calls.find(([name]) => name === 'selectModel'), [
'selectModel',
'session-one',
{ provider: 'openrouter', model: 'anthropic/claude-sonnet-4' },
{},
]);
});
test('/model rejects unknown IDs before creating a Session', async () => {
const { calls, harness, state } = fixture();
const result = await runModelCommand(
'/model DEEPSEEK-OFFICIAL/deepseek-v4-flash',
harness,
state,
'direct:one',
);
assert.match(result.message, /没有找到模型/);
assert.equal(calls.some(([name]) => name === 'createSession'), false);
assert.equal(calls.some(([name]) => name === 'selectModel'), false);
assert.match(
(await runModelCommand('/model missing-slash', harness, state, 'direct:one')).message,
/用法/,
);
});
test('/model creates and selects a blank Session before exposing its binding', async () => {
const { calls, harness, state, boundId } = fixture();
const signal = new AbortController().signal;
const result = await runModelCommand(
'/model deepseek-official/deepseek-v4-pro',
harness,
state,
'direct:one',
{ signal, control: 'control-one' },
);
assert.match(result.message, /模型已切换为/);
assert.equal(boundId(), 'session-created');
const operations = calls.map(([name]) => name);
assert.ok(operations.indexOf('listModels') < operations.indexOf('createSession'));
assert.ok(operations.indexOf('createSession') < operations.indexOf('selectModel'));
assert.ok(operations.indexOf('selectModel') < operations.indexOf('setSession'));
assert.deepEqual(calls.find(([name]) => name === 'selectModel'), [
'selectModel',
'session-created',
{ provider: 'deepseek-official', model: 'deepseek-v4-pro' },
{ signal },
]);
});
test('a failed first model selection leaves the conversation unbound', async () => {
const selectionError = new Error('provider unavailable');
selectionError.code = 'model-unavailable';
const { calls, harness, state, boundId } = fixture({ selectionError });
const result = await runModelCommand(
'/model deepseek-official/deepseek-v4-pro',
harness,
state,
'direct:one',
);
assert.match(result.message, /当前不可用|图片/);
assert.equal(boundId(), null);
assert.equal(calls.some(([name]) => name === 'createSession'), true);
assert.equal(calls.some(([name]) => name === 'selectModel'), true);
assert.equal(calls.some(([name]) => name === 'setSession'), false);
});
test('two concurrent first model switches share one created Session', async () => {
const { calls, harness, state, boundId } = fixture();
const results = await Promise.all([
runModelCommand(
'/model deepseek-official/deepseek-v4-pro',
harness,
state,
'direct:one',
),
runModelCommand(
'/model openrouter/anthropic/claude-sonnet-4',
harness,
state,
'direct:one',
),
]);
assert.ok(results.every(({ message }) => /模型已切换为/.test(message)));
assert.equal(boundId(), 'session-created');
assert.equal(calls.filter(([name]) => name === 'createSession').length, 1);
assert.equal(calls.filter(([name]) => name === 'setSession').length, 1);
assert.deepEqual(
calls.filter(([name]) => name === 'selectModel').map(([, sessionId]) => sessionId),
['session-created', 'session-created'],
);
});
test('a concurrent external binding is preserved after selecting an unbound Session', async () => {
const selectionStarted = deferred();
const releaseSelection = deferred();
const fixtureValue = fixture({
selectModelHook: async ({ sessionId }) => {
if (sessionId !== 'session-created') return;
selectionStarted.resolve();
await releaseSelection.promise;
},
});
const switching = runModelCommand(
'/model deepseek-official/deepseek-v4-pro',
fixtureValue.harness,
fixtureValue.state,
'direct:one',
);
await selectionStarted.promise;
await fixtureValue.state.setSession('direct:one', 'session-bound-elsewhere');
releaseSelection.resolve();
const result = await switching;
assert.match(result.message, /会话已发生变化.*重试/);
assert.equal(fixtureValue.boundId(), 'session-bound-elsewhere');
assert.equal(fixtureValue.calls.some((call) => (
call[0] === 'setSession' && call[2] === 'session-created'
)), false);
});
test('/model refuses pending interactions and active or running Sessions', async () => {
const pending = fixture({ initialSessionId: 'session-one' });
const pendingResult = await runModelCommand(
'/model deepseek-official/deepseek-v4-pro',
pending.harness,
pending.state,
'direct:one',
{ pendingInteraction: true },
);
assert.match(pendingResult.message, /等待你的回答或审批/);
assert.equal(pending.calls.some(([name]) => name === 'selectModel'), false);
for (const state of [{ running: true }, { activeTurn: true }]) {
const active = fixture({ initialSessionId: 'session-one', ...state });
const result = await runModelCommand(
'/model deepseek-official/deepseek-v4-pro',
active.harness,
active.state,
'direct:one',
{ control: 'owner-one' },
);
assert.match(result.message, /当前任务正在运行/);
assert.equal(active.calls.some(([name]) => name === 'models'), false);
assert.equal(active.calls.some(([name]) => name === 'selectModel'), false);
}
});
test('a missing bound Session is cleared and /models falls back to the global catalog', async () => {
const { calls, harness, state, boundId } = fixture({
initialSessionId: 'session-missing',
existing: false,
});
const result = await runModelCommand('/models', harness, state, 'direct:one');
assert.match(result.message, /deepseek-official\/deepseek-v4-flash/);
assert.equal(boundId(), null);
assert.equal(calls.some(([name]) => name === 'clearSession'), true);
assert.equal(calls.some(([name]) => name === 'listModels'), true);
});
test('model command failures use safe user-facing messages', async () => {
const privateError = new Error('provider leaked API key sk-private');
privateError.code = 'model-unavailable';
const selection = fixture({
initialSessionId: 'session-one',
selectionError: privateError,
});
const failed = await runModelCommand(
'/model deepseek-official/deepseek-v4-pro',
selection.harness,
selection.state,
'direct:one',
);
assert.match(failed.message, /当前不可用|图片/);
assert.doesNotMatch(failed.message, /sk-private/);
const listing = fixture({ globalCatalog: new Error('private endpoint') });
const unavailable = await runModelCommand('/models', listing.harness, listing.state, 'direct:one');
assert.match(unavailable.message, /暂时无法获取模型列表/);
assert.doesNotMatch(unavailable.message, /private endpoint/);
const bad = fixture({ globalCatalog: { groups: null, failures: [] } });
assert.match(
(await runModelCommand('/models', bad.harness, bad.state, 'direct:one')).message,
/暂时无法获取模型列表/,
);
});
test('non-model input is left for ordinary message routing', async () => {
assert.equal(await runModelCommand('hello', {}, {}, 'direct:one'), null);
});

View file

@ -17,6 +17,7 @@ import {
BotWorkspaceStore,
createBotWorkspaceScope,
} from '../src/channels/shared/bot-workspace-store.mjs';
import { runModelCommand } from '../src/channels/shared/model-command.mjs';
import { askInWorkspaceSession } from '../src/channels/shared/workspace-session.mjs';
async function fixture(t) {
@ -45,6 +46,20 @@ function deferred() {
return { promise, resolve };
}
async function within(promise, timeoutMs, message) {
let timer;
try {
return await Promise.race([
promise,
new Promise((_, reject) => {
timer = setTimeout(() => reject(new Error(message)), timeoutMs);
}),
]);
} finally {
clearTimeout(timer);
}
}
function memoryState(initial = {}) {
let sessions = { ...initial };
let clears = 0;
@ -188,6 +203,90 @@ test('the next message continues the bound Session without creating a new one',
assert.equal(createCalls, 0);
});
test('a first prompt and model switch share one binding without holding the lock during ask', async () => {
const state = memoryState();
const askStarted = deferred();
const releaseAsk = deferred();
const calls = [];
let creations = 0;
const catalog = {
groups: [{
id: 'provider',
name: 'Provider',
models: [{ id: 'model', name: 'Model' }],
}],
failures: [],
};
const harness = {
async createSession() {
creations += 1;
calls.push(['createSession']);
return `session-${creations}`;
},
workspaceSession(sessionId) {
return {
async sessionExists() { return true; },
async ask(text) {
calls.push(['ask', sessionId, text]);
askStarted.resolve();
await releaseAsk.promise;
return 'answer';
},
async isRunning() { return false; },
async hasActiveTurn() { return false; },
async models() {
return {
...catalog,
current: { provider: 'provider', model: 'model' },
routable: true,
};
},
async selectModel(selection) {
calls.push(['selectModel', sessionId, selection]);
return { selected: selection };
},
};
},
async listModels() { return catalog; },
};
const prompting = askInWorkspaceSession({
harness,
state,
key: 'conversation',
text: 'first prompt',
});
const switching = runModelCommand(
'/model provider/model',
harness,
state,
'conversation',
);
await askStarted.promise;
let switchResult;
try {
switchResult = await within(
switching,
500,
'model switching waited for the running ask instead of only the binding transaction',
);
} finally {
releaseAsk.resolve();
}
const promptResult = await prompting;
assert.match(switchResult.message, /模型已切换为/);
assert.deepEqual(promptResult, { sessionId: 'session-1', answer: 'answer' });
assert.equal(creations, 1);
assert.deepEqual(state.snapshot(), { conversation: 'session-1' });
assert.deepEqual(calls, [
['createSession'],
['ask', 'session-1', 'first prompt'],
['selectModel', 'session-1', { provider: 'provider', model: 'model' }],
]);
});
test('workspace sessions forward structured multimodal prompt content unchanged', async (t) => {
const { path, defaultWorkspace } = await fixture(t);
const workspaces = await new BotWorkspaceStore(path, { defaultWorkspace }).load();

View file

@ -21,7 +21,10 @@ import {
splitWorkspaceCommandMessage,
} from '../src/channels/shared/workspace-command.mjs';
import { TextHarnessBridge } from '../src/channels/shared/text-harness-bridge.mjs';
import { askInWorkspaceSession } from '../src/channels/shared/workspace-session.mjs';
import {
askInWorkspaceSession,
WORKSPACE_SESSION_STALE,
} from '../src/channels/shared/workspace-session.mjs';
import { HarnessClient as WeixinHarnessClient } from '../src/channels/weixin/harness-client.mjs';
import { HarnessClient as FeishuHarnessClient } from '../src/channels/feishu/harness-client.mjs';
import { HarnessClient as DingtalkHarnessClient } from '../src/channels/dingtalk/harness-client.mjs';
@ -205,6 +208,69 @@ test('an old session cannot be written back while RPC switches the bot workspace
assert.equal(existenceChecks, 0, 'stale sessions are rejected before asking Harness');
});
test('an old workspace session handle cannot list, select, stop, or steer after a switch', async (t) => {
const { path, defaultWorkspace, alternateWorkspace } = await fixture(t);
const workspaces = await new BotWorkspaceStore(path, { defaultWorkspace }).load();
await workspaces.ensure('bot_session_controls');
const targetCalls = [];
const harness = {
async getSessionModels(...args) { targetCalls.push(['models', ...args]); },
async selectSessionModel(...args) { targetCalls.push(['select', ...args]); },
async stopActiveTurn(...args) { targetCalls.push(['stop', ...args]); },
async steerActiveTurn(...args) { targetCalls.push(['steer', ...args]); },
};
const state = { async clearSessions() {} };
const scope = createBotWorkspaceScope(harness, {
botId: 'bot_session_controls', workspaces, state,
});
const oldSession = scope.harness.workspaceSession('session-old');
const controller = createWorkspaceAwareController({
status() { return { bots: [{ botId: 'bot_session_controls' }] }; },
}, { workspaces, stateFor: async () => state });
await controller.updateWorkspace('bot_session_controls', alternateWorkspace);
const control = { owner: {}, key: 'direct:one' };
for (const operation of [
() => oldSession.models(),
() => oldSession.selectModel({ provider: 'provider', model: 'model' }),
() => oldSession.stopActiveTurn(control),
() => oldSession.steerActiveTurn('continue', control),
]) {
await assert.rejects(operation(), (error) => error?.code === WORKSPACE_SESSION_STALE);
}
assert.deepEqual(targetCalls, []);
});
test('a control mutation that already started keeps its result across a workspace switch', async (t) => {
const { path, defaultWorkspace, alternateWorkspace } = await fixture(t);
const workspaces = await new BotWorkspaceStore(path, { defaultWorkspace }).load();
await workspaces.ensure('bot_started_controls');
const started = [];
let release;
const gate = new Promise((resolve) => { release = resolve; });
const harness = {
async stopActiveTurn() { started.push('stop'); await gate; return true; },
async steerActiveTurn() { started.push('steer'); await gate; return true; },
};
const state = { async clearSessions() {} };
const scope = createBotWorkspaceScope(harness, {
botId: 'bot_started_controls', workspaces, state,
});
const session = scope.harness.workspaceSession('session-old');
const control = { owner: {}, key: 'direct:one' };
const stop = session.stopActiveTurn(control);
const steer = session.steerActiveTurn('continue', control);
assert.deepEqual(started, ['stop', 'steer']);
const controller = createWorkspaceAwareController({
status() { return { bots: [{ botId: 'bot_started_controls' }] }; },
}, { workspaces, stateFor: async () => state });
await controller.updateWorkspace('bot_started_controls', alternateWorkspace);
release();
assert.deepEqual(await Promise.all([stop, steer]), [true, true]);
});
test('a prompt retries in the new workspace when switching after session creation', async (t) => {
const { path, defaultWorkspace, alternateWorkspace } = await fixture(t);
const workspaces = await new BotWorkspaceStore(path, { defaultWorkspace }).load();