fix(discord): close thread recovery races

src/channels/discord/discord-runtime.mjs

@@ -5,6 +5,7 @@ import { DiscordApi } from './discord-api.mjs';
 import { createDiscordBridgeStatus, DiscordHarnessBridge } from './discord-bridge.mjs';
 
 const DISCORD_GATEWAY_INTENTS = (1 << 0) | (1 << 9) | (1 << 12) | (1 << 15);
+const THREAD_RECOVERY_TIMEOUT_MS = 5_000;
 const RECONNECT_DELAYS_MS = Object.freeze([1_000, 3_000, 5_000, 10_000, 30_000]);
 const GUILD_TEXT = 0;
 const GUILD_ANNOUNCEMENT = 5;
@@ -87,13 +88,13 @@ function rememberChannel(channel, callback) {
 }
 
 function withConversationRoute(normalized, channel, botId, {
-  created = false,
+  fromSourceMessage = false,
   fallback = null,
   notice = null,
 } = {}) {
   const channelId = String(channel?.id ?? normalized.conversationId);
   const thread = isThreadChannel(channel);
-  const managed = thread && (created || String(channel?.owner_id ?? '') === String(botId));
+  const managed = thread && String(channel?.owner_id ?? '') === String(botId);
   const parentId = thread && channel?.parent_id ? String(channel.parent_id) : channelId;
   return {
     ...normalized,
@@ -102,7 +103,7 @@ function withConversationRoute(normalized, channel, botId, {
     requiresMention: thread ? !managed : normalized.kind === 'group',
     replyTarget: {
       channelId,
-      ...(!created && normalized.replyTarget?.replyToMessageId
+      ...(!fromSourceMessage && normalized.replyTarget?.replyToMessageId
         ? { replyToMessageId: normalized.replyTarget.replyToMessageId }
         : {}),
       ...(notice ? { notice } : {}),
@@ -239,6 +240,7 @@ export async function resolveDiscordMessageRoute(message, botId, {
 } = {}) {
   const normalized = normalizeDiscordMessage(message, botId, { fetchImpl });
   if (!normalized || normalized.senderIsBot) return normalized;
+  signal?.throwIfAborted();
   if (normalized.kind === 'direct') {
     return withConversationRoute(normalized, { id: normalized.conversationId, type: 1 }, botId);
   }
@@ -263,6 +265,7 @@ export async function resolveDiscordMessageRoute(message, botId, {
     });
   }
 
+  signal?.throwIfAborted();
   let created;
   try {
     created = rememberChannel(await api.startThreadFromMessage({
@@ -275,15 +278,21 @@ export async function resolveDiscordMessageRoute(message, botId, {
       throw new Error('Discord returned an invalid thread for the source message');
     }
   } catch (error) {
-    if (signal?.aborted) throw signal.reason ?? error;
     let recovered;
     try {
-      recovered = await findCreatedThread(api, message, { signal, onChannel });
+      recovered = await findCreatedThread(api, message, {
+        signal: AbortSignal.timeout(THREAD_RECOVERY_TIMEOUT_MS),
+        onChannel,
+      });
     } catch (recoveryError) {
+      if (signal?.aborted) throw signal.reason ?? error;
       await sendThreadUncertainNotice(api, normalized, signal);
       throw threadCreateUncertain(recoveryError);
     }
-    if (recovered) return withConversationRoute(normalized, recovered, botId, { created: true });
+    if (recovered) {
+      return withConversationRoute(normalized, recovered, botId, { fromSourceMessage: true });
+    }
+    if (signal?.aborted) throw signal.reason ?? error;
     if (uncertainThreadCreate(error)) {
       await sendThreadUncertainNotice(api, normalized, signal);
       throw threadCreateUncertain(error);
@@ -293,7 +302,7 @@ export async function resolveDiscordMessageRoute(message, botId, {
       notice: '无法创建 Thread，已直接在当前频道回复。',
     });
   }
-  return withConversationRoute(normalized, created, botId, { created: true });
+  return withConversationRoute(normalized, created, botId, { fromSourceMessage: true });
 }
 
 export class DiscordBotClient {

test/channels/discord/discord.test.mjs

@@ -671,6 +671,95 @@ test('Discord recovers an already-created thread after an uncertain or duplicate
   assert.equal(route.conversationRoute.threadId, message.id);
   assert.equal(route.conversationRoute.managed, true);
   assert.deepEqual(route.replyTarget, { channelId: message.id });
+
+  const foreignMessage = {
+    ...message,
+    id: '111111111111111151',
+    content: `<@${botId}> recover foreign thread`,
+  };
+  const foreign = await resolveDiscordMessageRoute(foreignMessage, botId, {
+    channel: { id: foreignMessage.channel_id, type: 0 },
+    api: {
+      async getChannel({ channelId }) {
+        assert.equal(channelId, foreignMessage.id);
+        return {
+          id: foreignMessage.id,
+          type: 11,
+          parent_id: foreignMessage.channel_id,
+          owner_id: '999999999999999999',
+        };
+      },
+      async startThreadFromMessage() {
+        const error = new Error('Thread already exists');
+        error.status = 400;
+        throw error;
+      },
+    },
+  });
+  assert.equal(foreign.addressed, true, 'the explicit source mention still starts this turn');
+  assert.equal(foreign.requiresMention, true);
+  assert.equal(foreign.conversationRoute.managed, false);
+  assert.deepEqual(foreign.replyTarget, { channelId: foreignMessage.id });
+});
+
+test('Discord converges a post-dispatch abort but sends no create request after a pre-dispatch abort', async () => {
+  const botId = '1234567890123456789';
+  const base = {
+    channel_id: '222222222222222252',
+    guild_id: '444444444444444444',
+    author: { id: '333333333333333333', bot: false },
+    mentions: [{ id: botId }],
+    content: `<@${botId}> abort boundary`,
+  };
+  const before = new AbortController();
+  const beforeReason = new DOMException('cancelled before dispatch', 'AbortError');
+  before.abort(beforeReason);
+  let preDispatchCalls = 0;
+  await assert.rejects(() => resolveDiscordMessageRoute({
+    ...base,
+    id: '111111111111111152',
+  }, botId, {
+    channel: { id: base.channel_id, type: 0 },
+    signal: before.signal,
+    api: {
+      async getChannel() { preDispatchCalls += 1; },
+      async startThreadFromMessage() { preDispatchCalls += 1; },
+    },
+  }), (error) => error === beforeReason);
+  assert.equal(preDispatchCalls, 0);
+
+  const after = new AbortController();
+  const afterReason = new DOMException('cancelled after dispatch', 'AbortError');
+  const message = { ...base, id: '111111111111111153' };
+  let starts = 0;
+  let lookups = 0;
+  const recovered = await resolveDiscordMessageRoute(message, botId, {
+    channel: { id: message.channel_id, type: 0 },
+    signal: after.signal,
+    api: {
+      async startThreadFromMessage() {
+        starts += 1;
+        after.abort(afterReason);
+        throw afterReason;
+      },
+      async getChannel({ channelId, signal }) {
+        lookups += 1;
+        assert.equal(channelId, message.id);
+        assert.notEqual(signal, after.signal);
+        assert.equal(signal.aborted, false);
+        return {
+          id: message.id,
+          type: 11,
+          parent_id: message.channel_id,
+          owner_id: botId,
+        };
+      },
+    },
+  });
+  assert.equal(starts, 1);
+  assert.equal(lookups, 1);
+  assert.equal(recovered.conversationId, message.id);
+  assert.equal(recovered.conversationRoute.managed, true);
 });
 
 test('Discord falls back before routing when the parent is unsupported or thread creation is rejected', async () => {
@@ -794,6 +883,67 @@ test('Discord merges a deterministic Thread fallback notice into one delivered a
   assert.equal(edits[0].content, `${streamTarget.notice}\n\nstreamed answer`);
 });
 
+test('Discord keeps streamed text and result files on the final created Thread target', async () => {
+  const botId = '1234567890123456789';
+  const message = {
+    id: '111111111111111165',
+    channel_id: '222222222222222265',
+    guild_id: '444444444444444444',
+    author: { id: '333333333333333333', bot: false },
+    mentions: [{ id: botId }],
+    content: `<@${botId}> deliver in thread`,
+  };
+  const operations = [];
+  const api = {
+    async getChannel() {
+      assert.fail('the cached parent channel must be reused');
+    },
+    async startThreadFromMessage() {
+      return {
+        id: message.id,
+        type: 11,
+        parent_id: message.channel_id,
+        owner_id: botId,
+      };
+    },
+    async createMessage(options) {
+      operations.push({ operation: 'create', ...options });
+      return { id: '888888888888888881' };
+    },
+    async editMessage(options) {
+      operations.push({ operation: 'edit', ...options });
+      return { id: options.messageId };
+    },
+    async createFileMessage(options) {
+      operations.push({ operation: 'file', ...options });
+      return { id: '888888888888888882' };
+    },
+    async sendTyping(options) {
+      operations.push({ operation: 'typing', ...options });
+    },
+  };
+  const route = await resolveDiscordMessageRoute(message, botId, {
+    channel: { id: message.channel_id, type: 0 },
+    api,
+  });
+  const client = new DiscordBotClient({ api });
+  await client.sendTyping(route.replyTarget);
+  const stream = await client.openStream(route.replyTarget);
+  await stream.finish('thread final answer');
+  await client.sendFile(route.replyTarget, {
+    fileName: 'result.txt',
+    bytes: Buffer.from('thread result'),
+  });
+
+  assert.deepEqual(operations.map(({ operation }) => operation), [
+    'typing', 'create', 'edit', 'file',
+  ]);
+  assert.equal(operations.every(({ channelId }) => channelId === message.id), true);
+  assert.equal(operations.some(({ channelId }) => channelId === message.channel_id), false);
+  assert.equal(operations.find(({ operation }) => operation === 'create').replyToMessageId, undefined);
+  assert.equal(operations.find(({ operation }) => operation === 'file').replyToMessageId, undefined);
+});
+
 test('Discord keeps image attachments in images and exposes ordinary attachments as files', async () => {
   const png = Buffer.from([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]);
   const calls = [];
@@ -967,7 +1117,7 @@ test('Discord runtime keeps one isolated Session per managed thread and reuses i
     hasSeen: (messageId) => seen.has(messageId),
     async markSeen(messageId) { seen.add(messageId); },
   };
-  const runtime = new DiscordRuntime({
+  const createRuntime = () => new DiscordRuntime({
     config: { botId: 'discord_test', platformId: botId, name: 'Harness Discord' },
     token: TOKEN,
     harness,
@@ -983,6 +1133,7 @@ test('Discord runtime keeps one isolated Session per managed thread and reuses i
     random: () => 0.5,
     logger: { warn() {}, error(...args) { assert.fail(args.join(' ')); } },
   });
+  let runtime = createRuntime();
   await runtime.start();
   socket.emit('message', {
     data: JSON.stringify({
@@ -1133,6 +1284,50 @@ test('Discord runtime keeps one isolated Session per managed thread and reuses i
   );
   assert.equal(sessions.has(`group:${parentChannelId}`), false);
   await runtime.stop();
+
+  runtime = createRuntime();
+  await runtime.start();
+  socket.emit('message', {
+    data: JSON.stringify({
+      op: 0,
+      t: 'GUILD_CREATE',
+      s: 20,
+      d: {
+        id: '444444444444444444',
+        channels: [{ id: parentChannelId, type: 0 }],
+        threads: [{
+          id: firstMessage.id,
+          type: 11,
+          parent_id: parentChannelId,
+          owner_id: botId,
+        }],
+      },
+    }),
+  });
+  socket.emit('message', {
+    data: JSON.stringify({
+      op: 0,
+      t: 'MESSAGE_CREATE',
+      s: 21,
+      d: {
+        id: '111111111111111174',
+        channel_id: firstMessage.id,
+        guild_id: '444444444444444444',
+        author: { id: '333333333333333330', bot: false },
+        mentions: [],
+        content: 'continue after restart',
+      },
+    }),
+  });
+  await eventually(() => runtime.status.messagesReplied === 1);
+  assert.deepEqual(asks[5], {
+    sessionId: 'session-1',
+    text: 'continue after restart',
+    files: 0,
+  });
+  assert.equal(starts.length, 3, 'restart must reuse the persisted bot-owned Thread');
+  assert.equal(sessions.get(`group:${firstMessage.id}`), 'session-1');
+  await runtime.stop();
 });
 
 test('Discord runtime records one uncertain Thread result and suppresses Gateway replays', async () => {