import { createCipheriv, createHash, randomBytes } from "node:crypto";
import { mkdtemp, rm, writeFile } from "node:fs/promises";
import { join } from "node:path";
import { tmpdir } from "node:os";
import { Readable } from "node:stream";
import { ApiServerChannel } from "./src/adapters/channel/api-server/ApiServerChannel.js";
import { ApiServerSessionMapper } from "./src/adapters/channel/api-server/ApiServerSessionMapper.js";
import { WebhookChannel } from "./src/adapters/channel/webhook/WebhookChannel.js";
import { WebhookSessionMapper } from "./src/adapters/channel/webhook/WebhookSessionMapper.js";
import { WeComCallbackChannel } from "./src/adapters/channel/wecom-callback/WeComCallbackChannel.js";
import { WeComCallbackSessionMapper } from "./src/adapters/channel/wecom-callback/WeComCallbackSessionMapper.js";
import { WeixinChannel } from "./src/adapters/channel/weixin/WeixinChannel.js";
import { WeixinSessionMapper } from "./src/adapters/channel/weixin/WeixinSessionMapper.js";
import type { Gateway, GatewayEvent } from "./src/gateway/index.js";
class Deferred<T> {
readonly promise: Promise<T>;
private resolvePromise!: (value: T) => void;
constructor() {
this.promise = new Promise<T>((resolve) => { this.resolvePromise = resolve; });
}
resolve(value: T): void { this.resolvePromise(value); }
}
type SubmitCall = { sessionKey: string; channelKey: string; message: string };
class BlockingGateway {
readonly calls: SubmitCall[] = [];
readonly abortCalls: Array<{ sessionKey: string; runId?: string; reason?: string }> = [];
readonly holdStarted = new Deferred<void>();
readonly release = new Deferred<void>();
async *submitTurn(input: { sessionKey: string; channelKey: string; message: string }): AsyncIterable<GatewayEvent> {
this.calls.push({ sessionKey: input.sessionKey, channelKey: input.channelKey, message: input.message });
yield { type: "turn_started", runId: "run-hold" };
if (input.message === "hold") {
this.holdStarted.resolve();
await this.release.promise;
yield { type: "error", code: "agent_aborted", message: "fixture aborted" } as GatewayEvent;
return;
}
yield { type: "assistant_text_delta", text: "fixture reply" };
}
async abortTurn(input: { sessionKey: string; runId?: string; reason?: string }): Promise<void> {
this.abortCalls.push({ ...input });
this.release.resolve();
}
}
function gatewayForFixture(gateway: BlockingGateway): Gateway {
return gateway as unknown as Gateway;
}
async function waitFor(predicate: () => boolean, label: string, timeoutMs = 3000): Promise<void> {
const deadline = Date.now() + timeoutMs;
while (!predicate()) {
if (Date.now() >= deadline) throw new Error(`fixture timeout waiting for ${label}`);
await new Promise((resolve) => setTimeout(resolve, 5));
}
}
function noOpLogger() {
return { info() {}, warn() {}, error() {} };
}
function holdMessage(fromUserId: string, text: string): any {
return {
message_type: 1,
from_user_id: fromUserId,
context_token: "fixture-context",
item_list: [{ type: 1, text_item: { text } }],
};
}
async function runWeixin(): Promise<Record<string, unknown>> {
const tempDir = await mkdtemp(join(tmpdir(), "pilotdeck-witness-new-"));
const credentialsPath = join(tempDir, "credentials.json");
await writeFile(credentialsPath, JSON.stringify({
baseUrl: "https://fixture.invalid",
botToken: "fixture-token",
accountId: "fixture-account",
cursor: "fixture-cursor",
}));
const gateway = new BlockingGateway();
const userId = "weixin-user";
const mapper = new WeixinSessionMapper({
activeByChatId: { [userId]: "weixin:chat=weixin-user:s_old" },
projectByChatId: {},
}, () => "weixin-new");
const secondPoll = new Deferred<any>();
const stopPoll = new Deferred<any>();
let pollCount = 0;
const sentReplies: string[] = [];
const client = {
cursor: "fixture-cursor",
poll: async () => {
pollCount++;
if (pollCount === 1) return { ret: 0, msgs: [holdMessage(userId, "hold")] };
if (pollCount === 2) return secondPoll.promise;
return stopPoll.promise;
},
sendTextChunked: async (_to: string, text: string) => { sentReplies.push(text); return 1; },
sendMedia: async () => undefined,
getUploadUrl: async () => ({}),
sendTyping: async () => undefined,
};
const channel = new WeixinChannel({
credentialsPath,
mapper,
clientFactory: () => client,
});
const handle = await channel.start({ gateway: gatewayForFixture(gateway), logger: noOpLogger() });
await gateway.holdStarted.promise;
secondPoll.resolve({ ret: 0, msgs: [holdMessage(userId, "/new")] });
await waitFor(() => gateway.abortCalls.length === 1, "Weixin abortTurn from /new");
await waitFor(() => sentReplies.includes("已创建新会话。"), "Weixin /new acknowledgement");
const snapshot = mapper.snapshot();
gateway.release.resolve();
await waitFor(() => !(channel as any).activeChats.has(userId), "Weixin active turn cleanup");
stopPoll.resolve({ ret: 0, msgs: [] });
await handle.stop("fixture cleanup");
await rm(tempDir, { recursive: true, force: true });
return {
entry: "WeixinChannel.start -> fake iLink poll -> dispatchMessage",
initialMessage: "hold",
commandMessage: "/new",
submitCalls: gateway.calls,
abortCalls: gateway.abortCalls,
mapperSnapshot: snapshot,
acknowledgementSent: sentReplies.includes("已创建新会话。"),
};
}
async function runApiServer(): Promise<Record<string, unknown>> {
const gateway = new BlockingGateway();
const chatId = "api-chat";
const mapper = new ApiServerSessionMapper({
activeByChatId: { [chatId]: "api_server:chat=api-chat:s_old" },
}, () => "api-new");
const channel = new ApiServerChannel({ port: 8642, mapper });
(channel as any).gateway = gatewayForFixture(gateway);
(channel as any).logger = noOpLogger();
const firstResponse = mockResponse();
const firstRequest = (channel as any).handleChatCompletions(
bodyRequest({ messages: [{ role: "user", content: "hold" }] }, chatId),
firstResponse,
);
await gateway.holdStarted.promise;
const commandResponse = mockResponse();
await (channel as any).handleChatCompletions(
bodyRequest({ messages: [{ role: "user", content: "/new" }] }, chatId),
commandResponse,
);
const snapshotWhileActive = mapper.snapshot();
gateway.release.resolve();
await firstRequest;
return {
entry: "ApiServerChannel.start -> POST /v1/chat/completions",
initialMessage: "hold",
commandMessage: "/new",
commandHttp: { status: commandResponse.statusCode, body: JSON.parse(commandResponse.body) },
firstHttpStatus: firstResponse.statusCode,
submitCalls: gateway.calls,
abortCalls: gateway.abortCalls,
mapperSnapshotWhileActive: snapshotWhileActive,
};
}
function encryptWeComXml(xml: string, encodingAesKey: string, corpId: string): string {
const key = Buffer.from(`${encodingAesKey}=`, "base64");
const body = Buffer.from(xml, "utf8");
const corp = Buffer.from(corpId, "utf8");
const length = Buffer.alloc(4);
length.writeUInt32BE(body.length, 0);
const unpadded = Buffer.concat([randomBytes(16), length, body, corp]);
const padLength = 32 - (unpadded.length % 32);
const padded = Buffer.concat([unpadded, Buffer.alloc(padLength, padLength)]);
const cipher = createCipheriv("aes-256-cbc", key, key.subarray(0, 16));
cipher.setAutoPadding(false);
return Buffer.concat([cipher.update(padded), cipher.final()]).toString("base64");
}
function signedWeComRequest(text: string, token: string, encodingAesKey: string, corpId: string): { path: string; body: string } {
const timestamp = "1700000000";
const nonce = "fixture-nonce";
const plain = `<xml><ToUserName><![CDATA[${corpId}]]></ToUserName><FromUserName><![CDATA[wecom-user]]></FromUserName><CreateTime>1700000000</CreateTime><MsgType><![CDATA[text]]></MsgType><Content><![CDATA[${text}]]></Content></xml>`;
const encrypted = encryptWeComXml(plain, encodingAesKey, corpId);
const signature = createHash("sha1").update([token, timestamp, nonce, encrypted].sort().join("")).digest("hex");
return {
path: `/callback?msg_signature=${encodeURIComponent(signature)}×tamp=${timestamp}&nonce=${encodeURIComponent(nonce)}`,
body: `<xml><Encrypt><![CDATA[${encrypted}]]></Encrypt></xml>`,
};
}
async function runWeComCallback(): Promise<Record<string, unknown>> {
const gateway = new BlockingGateway();
const corpId = "fixture-corp";
const token = "fixture-callback-token";
const encodingAesKey = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA";
const mapper = new WeComCallbackSessionMapper({
activeByChatId: { "wecom-user": "wecom_callback:chat=wecom-user:s_old" },
}, () => "wecom-new");
const channel = new WeComCallbackChannel({
corpId,
agentId: "1",
secret: "fixture-secret",
token,
encodingAesKey,
mapper,
});
(channel as any).gateway = gatewayForFixture(gateway);
(channel as any).logger = noOpLogger();
const send = async (text: string) => {
const request = signedWeComRequest(text, token, encodingAesKey, corpId);
const response = mockCallbackResponse();
await (channel as any).onHttp(
bodyRequest(request.body, "", request.path),
response,
);
return { status: response.statusCode, body: response.body };
};
const firstResponse = await send("hold");
await gateway.holdStarted.promise;
const commandResponse = await send("/new");
const snapshotWhileActive = mapper.snapshot();
gateway.release.resolve();
await waitFor(() => !(channel as any).activeChats.has("wecom-user"), "WeCom active turn cleanup");
return {
entry: "WeComCallbackChannel.start -> signed encrypted POST /callback",
initialMessage: "hold",
commandMessage: "/new",
firstHttp: firstResponse,
commandHttp: commandResponse,
submitCalls: gateway.calls,
abortCalls: gateway.abortCalls,
mapperSnapshotWhileActive: snapshotWhileActive,
};
}
async function runWebhook(): Promise<Record<string, unknown>> {
const gateway = new BlockingGateway();
const holdChatId = "webhook:fixture:hold-1";
const mapper = new WebhookSessionMapper({
activeByChatId: { [holdChatId]: "webhook:chat=webhook:fixture:hold-1:s_old" },
}, () => "webhook-new");
const channel = new WebhookChannel({
port: 8643,
routes: { fixture: { secret: "__INSECURE_NO_AUTH__", deliver: "log" } },
mapper,
});
(channel as any).gateway = gatewayForFixture(gateway);
(channel as any).logger = noOpLogger();
const send = async (text: string, deliveryId: string) => {
const response = mockResponse();
await (channel as any).handleWebhook(
bodyRequest({ text, sender: "stable-sender" }, "", "/webhooks/fixture", {
"x-delivery-id": deliveryId,
}),
response,
"fixture",
);
return { status: response.statusCode, body: JSON.parse(response.body) };
};
const firstResponse = await send("hold", "hold-1");
await gateway.holdStarted.promise;
const duplicateCommandResponse = await send("/new", "hold-1");
const distinctCommandResponse = await send("/new", "new-1");
await waitFor(() => Boolean(mapper.snapshot().activeByChatId["webhook:fixture:new-1"]), "Webhook distinct /new mapping");
const snapshotWhileActive = mapper.snapshot();
gateway.release.resolve();
await waitFor(() => !(channel as any).activeChats.has(holdChatId), "Webhook active turn cleanup");
return {
entry: "WebhookChannel.start -> POST /webhooks/fixture",
initialMessage: "hold",
commandMessage: "/new",
firstHttp: firstResponse,
duplicateCommandHttp: duplicateCommandResponse,
distinctCommandHttp: distinctCommandResponse,
submitCalls: gateway.calls,
abortCalls: gateway.abortCalls,
mapperSnapshotWhileActive: snapshotWhileActive,
};
}
function bodyRequest(body: unknown, sessionId = "", url = "/", extraHeaders: Record<string, string> = {}): any {
const payload = typeof body === "string" ? body : JSON.stringify(body);
const request = Readable.from([Buffer.from(payload)]) as any;
request.method = "POST";
request.url = url;
request.headers = {
host: "fixture.local",
"content-type": "application/json",
...(sessionId ? { "x-hermes-session-id": sessionId } : {}),
...extraHeaders,
};
return request;
}
function mockResponse(): any {
return {
statusCode: 200,
headers: {} as Record<string, string>,
body: "",
chunks: [] as string[],
setHeader(name: string, value: string) { this.headers[name.toLowerCase()] = String(value); },
flushHeaders() {},
write(chunk: string | Buffer) { this.chunks.push(String(chunk)); return true; },
end(chunk?: string | Buffer) {
if (chunk != null) this.chunks.push(String(chunk));
this.body = this.chunks.join("");
this.ended = true;
},
};
}
function mockCallbackResponse(): any {
const response = mockResponse();
response.writeHead = (statusCode: number, headers?: Record<string, string>) => {
response.statusCode = statusCode;
if (headers) Object.assign(response.headers, headers);
return response;
};
return response;
}
const actual = {
weixin: await runWeixin(),
api_server: await runApiServer(),
wecom_callback: await runWeComCallback(),
webhook: await runWebhook(),
};
console.log(JSON.stringify({
candidateId: "cand-command-active-new-divergence",
entryId: "entry-channel-interactive-controls",
actual,
}, null, 2));
Summary
The API returns HTTP 429 session_busy before mapper resolution. WeCom reports provider delivery success while dropping /new without rotation or abort. Discord silently drops bare and suffixed /new while the old turn remains active and later emits its old reply. Idle controls reach the reset branches.
Expected behavior
For an active logical chat, /new should rotate the mapper, abort and fence the old run, and acknowledge bare /new or forward exactly one suffix turn on the fresh session.
Actual behavior
The API returns HTTP 429 session_busy before mapper resolution. WeCom reports provider delivery success while dropping /new without rotation or abort. Discord silently drops bare and suffixed /new while the old turn remains active and later emits its old reply. Idle controls reach the reset branches.
Impact
Users cannot reliably reset or cancel active conversations across adapters, and some adapters provide no recoverable user-visible result.
Reproduction
Start an active turn and send /new through the API, WeCom callback, and Discord adapters, both bare and with a suffix. Record mapper rotation, abort/fence calls, provider acknowledgement, and any later output from the old turn. An active reset should have one consistent reset/acknowledgement contract; the observed result is API session_busy, WeCom acknowledgement while dropping /new, and Discord silently dropping /new while the old reply later arrives.
Minimal reproduction script
From the repository root, save this as repro_active_new_divergence.mts and run:
pnpm install --frozen-lockfile pnpm exec tsx repro_active_new_divergence.mtsRelevant source locations
src/adapters/channel/api-server/ApiServerChannel.ts:238-285src/adapters/channel/wecom-callback/WeComCallbackChannel.ts:175-253src/adapters/channel/discord/DiscordChannel.ts:138-194src/adapters/channel/protocol/ChannelCommandRegistry.ts:65-72src/adapters/channel/wecom-callback/WeComCallbackChannel.ts:236-245src/adapters/channel/weixin/WeixinChannel.ts:413-427Suggested direction
Make the external-input path establish one durable, identity-bound state/receipt before returning success; propagate explicit terminal outcomes to every channel and client; and add a regression test for the reproduced boundary.
This report is about functional behavior, not security. The reproduction uses deterministic in-memory or isolated fixtures and contains no credentials or private data.