feat(runtime-host): own bot conversation continuity

apps/desktop/src/main/__tests__/bot-incoming-project-cwd.test.ts

@@ -21,6 +21,7 @@ import { strict as assert } from 'node:assert';
 import { describe, it } from 'node:test';
 import type { BotIncomingMessage, BotRegistry } from '@maka/runtime/bots';
 import { createBotIncomingMainService } from '../bot-incoming-main.js';
+import { createTestBotSessionAdapter } from './bot-session-adapter-fixture.js';
 
 describe('bot incoming new-session cwd', () => {
   it('leaves the cwd to the shared desktop session resolver', async () => {
@@ -35,32 +36,31 @@ describe('bot incoming new-session cwd', () => {
           return true;
         },
       } as unknown as BotRegistry,
-      sessions: {
-        async createSession(input) {
+      sessions: createTestBotSessionAdapter({
+        async resolveSession(input) {
           createInput = input;
-          throw new Error('__short_circuit_after_create__');
-        },
-        async prepareSession() {
-          throw new Error('prepareSession must not be reached');
+          return { kind: 'permission_refused' };
         },
         async runTurn() {
           throw new Error('runTurn must not be reached');
         },
-      },
+      }),
     });
 
     await service.handleBotIncomingMessage({
       platform: 'telegram',
       userId: 'u',
       userName: 'U',
-      chatId: 'c1',
+      conversationId: 'c1',
+      sourceEventId: 'source-1',
+      replyTarget: { chatId: 'c1', replyToMessageId: 'source-1' },
       isGroup: false,
       text: 'hello',
-      sourceMessageId: '',
       receivedAt: Date.now(),
     } as unknown as BotIncomingMessage);
 
     assert.deepEqual(createInput, {
+      conversationId: 'telegram:c1',
       name: 'Telegram 任务',
       labels: ['bot', 'telegram'],
     });

apps/desktop/src/main/__tests__/bot-incoming-session-lifecycle.test.ts

@@ -18,72 +18,162 @@
  */
 
 import assert from 'node:assert/strict';
-import { describe, test } from 'node:test';
-import type { BotIncomingMessage, BotRegistry } from '@maka/runtime/bots';
+import test from 'node:test';
+import type { BotIncomingMessage } from '@maka/runtime/bots';
 import { createBotIncomingMainService } from '../bot-incoming-main.js';
-import { BotSessionUnavailableError } from '../bot-session-adapter.js';
+import { createTestBotSessionAdapter } from './bot-session-adapter-fixture.js';
 
-async function waitFor(predicate: () => boolean): Promise<void> {
-  const deadline = Date.now() + 1_000;
-  while (!predicate()) {
-    if (Date.now() >= deadline) throw new Error('Timed out waiting for bot lifecycle test');
-    await new Promise<void>((resolve) => setImmediate(resolve));
-  }
-}
+test('routes one external conversation through Host resolution and stable message admission', async () => {
+  const resolutions: unknown[] = [];
+  const turns: unknown[] = [];
+  const sent: string[] = [];
+  const sessions = createTestBotSessionAdapter({
+    async resolveSession(input) {
+      resolutions.push(input);
+      return { kind: 'ready', sessionId: 'session-1' };
+    },
+    async runTurn(input) {
+      if (input.admissionMode === 'replay_only') return { kind: 'admission_required' };
+      turns.push(input);
+      return { kind: 'completed', text: 'reply' };
+    },
+  });
+  const service = createBotIncomingMainService({
+    sessions,
+    botRegistry: registry(sent),
+  });
 
-describe('bot session lifecycle bindings', () => {
-  test('rebinds a conversation after its archived session rejects a send', async () => {
-    const created: string[] = [];
-    const sent: string[] = [];
-    const replies: string[] = [];
-    let ensureCalls = 0;
-    const sessions = {
-      async createSession() {
-        const id = `bot-session-${created.length + 1}`;
-        created.push(id);
-        return id;
-      },
-      async prepareSession(sessionId: string) {
-        ensureCalls += 1;
-        if (sessionId === 'bot-session-1' && ensureCalls === 1) {
-          throw new BotSessionUnavailableError('archived');
-        }
-        return 'ready' as const;
-      },
-      async runTurn({ sessionId }: { sessionId: string }) {
-        sent.push(sessionId);
-        return { kind: 'completed' as const, text: `reply from ${sessionId}` };
-      },
-    };
+  await service.handleBotIncomingMessage(message({ sourceEventId: 'source-1', text: 'first' }));
+  await service.handleBotIncomingMessage(message({ sourceEventId: 'source-2', text: 'second' }));
 
-    const service = createBotIncomingMainService({
-      botRegistry: {
-        async sendMessage(_platform: string, _chatId: string, text: string) {
-          replies.push(text);
-          return 'message-id';
-        },
-        async sendTypingIndicator() {
-          return true;
+  assert.deepEqual(
+    resolutions.map((entry) => (entry as { conversationId: string }).conversationId),
+    ['telegram:chat-1', 'telegram:chat-1'],
+  );
+  assert.equal(turns.length, 2);
+  assert.equal((turns[0] as { messageId: string }).messageId.length, 68);
+  assert.notEqual(
+    (turns[0] as { messageId: string }).messageId,
+    (turns[1] as { messageId: string }).messageId,
+  );
+  assert.deepEqual(sent, ['reply', 'reply']);
+  await service.close();
+});
+
+test('routes every source delivery to Host idempotency with the same stable message id', async () => {
+  const firstIds: string[] = [];
+  const event = message({ sourceEventId: 'stable-source' });
+  const create = (ids: string[]) =>
+    createBotIncomingMainService({
+      sessions: createTestBotSessionAdapter({
+        async runTurn(input) {
+          ids.push(input.messageId);
+          return { kind: 'completed', text: 'reply' };
         },
-      } as unknown as BotRegistry,
-      sessions,
+      }),
+      botRegistry: registry([]),
     });
 
-    const base = {
-      platform: 'telegram',
-      userId: 'user',
-      userName: 'User',
-      chatId: 'chat',
-      isGroup: false,
-      receivedAt: Date.now(),
-    };
-    await service.handleBotIncomingMessage({ ...base, text: 'first', sourceMessageId: 'source-1' } as BotIncomingMessage);
-    await waitFor(() => replies.length === 1);
-    await service.handleBotIncomingMessage({ ...base, text: 'second', sourceMessageId: 'source-2', receivedAt: Date.now() + 1 } as BotIncomingMessage);
-    await waitFor(() => replies.length === 2);
+  const first = create(firstIds);
+  await first.handleBotIncomingMessage(event);
+  await first.handleBotIncomingMessage(event);
+  assert.equal(firstIds.length, 2);
+  assert.equal(firstIds[0], firstIds[1]);
+  await first.close();
+
+  const successorIds: string[] = [];
+  const successor = create(successorIds);
+  await successor.handleBotIncomingMessage(event);
+  assert.deepEqual(successorIds, [firstIds[0]]);
+  await successor.close();
+});
+
+test('service recreation replays more than the transient burst before admitting new work', async () => {
+  const admissions: Array<{ messageId: string; admissionMode: string }> = [];
+  const sent: string[] = [];
+  let replayProbeCount = 0;
+  const service = createBotIncomingMainService({
+    sessions: createTestBotSessionAdapter({
+      async runTurn(input) {
+        admissions.push({
+          messageId: input.messageId,
+          admissionMode: input.admissionMode ?? 'allow',
+        });
+        if (input.admissionMode === 'replay_only') {
+          const index = replayProbeCount++;
+          return index < 9
+            ? { kind: 'completed', text: `reply-${index}` }
+            : { kind: 'admission_required' };
+        }
+        return { kind: 'completed', text: 'new-reply' };
+      },
+    }),
+    botRegistry: registry(sent),
+  });
+
+  for (let index = 0; index < 9; index++) {
+    await service.handleBotIncomingMessage(
+      message({ sourceEventId: `retry-${index}`, text: `retry-${index}` }),
+    );
+  }
+  await service.handleBotIncomingMessage(
+    message({ sourceEventId: 'new-source', text: 'new', conversationId: 'new-chat' }),
+  );
+
+  assert.equal(admissions.filter((entry) => entry.admissionMode === 'replay_only').length, 10);
+  assert.equal(admissions.filter((entry) => entry.admissionMode === 'allow').length, 1);
+  assert.deepEqual(sent, [
+    ...Array.from({ length: 9 }, (_, index) => `reply-${index}`),
+    'new-reply',
+  ]);
+  await service.close();
+});
 
-    assert.deepEqual(created, ['bot-session-1', 'bot-session-2']);
-    assert.deepEqual(sent, ['bot-session-1', 'bot-session-2']);
-    assert.deepEqual(replies, ['reply from bot-session-1', 'reply from bot-session-2']);
+test('releases a direct-message binding through a source-correlated reset operation', async () => {
+  const releases: unknown[] = [];
+  const sent: string[] = [];
+  const service = createBotIncomingMainService({
+    sessions: createTestBotSessionAdapter({
+      async releaseConversation(input) {
+        releases.push(input);
+        return true;
+      },
+    }),
+    botRegistry: registry(sent),
   });
+
+  await service.handleBotIncomingMessage(message({ text: 'reset', sourceEventId: 'reset-1' }));
+
+  assert.equal(releases.length, 1);
+  assert.equal((releases[0] as { conversationId: string }).conversationId, 'telegram:chat-1');
+  assert.match((releases[0] as { operationId: string }).operationId, /^bot_[a-f0-9]{64}$/);
+  assert.deepEqual(sent, ['任务已重置，下一条消息会开新任务。']);
+  await service.close();
 });
+
+function message(overrides: Partial<BotIncomingMessage> = {}): BotIncomingMessage {
+  return {
+    platform: 'telegram',
+    userId: 'user-1',
+    userName: 'Alice',
+    conversationId: 'chat-1',
+    sourceEventId: 'source-1',
+    replyTarget: { chatId: 'chat-1', replyToMessageId: 'source-1' },
+    isGroup: false,
+    text: 'hello',
+    receivedAt: 1,
+    ...overrides,
+  };
+}
+
+function registry(sent: string[]) {
+  return {
+    async sendMessage(_platform: string, _chatId: string, text: string) {
+      sent.push(text);
+      return 'sent';
+    },
+    async sendTypingIndicator() {
+      return true;
+    },
+  } as never;
+}

apps/desktop/src/main/__tests__/bot-incoming-typing.test.ts

@@ -22,6 +22,7 @@ import { getEventListeners } from 'node:events';
 import { test } from 'node:test';
 import type { BotIncomingMessage, BotRegistry } from '@maka/runtime/bots';
 import { createBotIncomingMainService } from '../bot-incoming-main.js';
+import { createTestBotSessionAdapter } from './bot-session-adapter-fixture.js';
 
 async function waitFor(predicate: () => boolean, message: string): Promise<void> {
   for (let attempt = 0; attempt < 100; attempt += 1) {
@@ -60,28 +61,24 @@ test('the bot typing loop owns only its active abort listener', async (t) => {
         throw new Error('typing unavailable');
       },
     } as unknown as BotRegistry,
-    sessions: {
-      async createSession() {
-        return 'bot-session';
-      },
-      async prepareSession() {
-        return 'ready';
-      },
-      async runTurn() {
+    sessions: createTestBotSessionAdapter({
+      async runTurn(input) {
+        if (input.admissionMode === 'replay_only') return { kind: 'admission_required' };
         await turnReleased;
         return { kind: 'completed', text: 'Bot reply' };
       },
-    },
+    }),
   });
 
   const handling = service.handleBotIncomingMessage({
     platform: 'telegram',
     userId: 'user',
     userName: 'User',
-    chatId: 'chat',
+    conversationId: 'chat',
+    sourceEventId: 'source',
+    replyTarget: { chatId: 'chat', replyToMessageId: 'source' },
     isGroup: false,
     text: 'hello',
-    sourceMessageId: 'source',
     receivedAt: Date.now(),
   } as BotIncomingMessage);
 
@@ -125,7 +122,7 @@ test('streams reply snapshots and persists the final reply through one channel s
         options: { isGroup: boolean; streamId: string },
       ) {
         assert.equal(options.isGroup, false);
-        assert.match(options.streamId, /^[0-9a-f-]{36}$/);
+        assert.match(options.streamId, /^bot_[0-9a-f]{64}$/);
         return {
           update(text: string) {
             updates.push(text);
@@ -145,29 +142,25 @@ test('streams reply snapshots and persists the final reply through one channel s
         return true;
       },
     } as unknown as BotRegistry,
-    sessions: {
-      async createSession() {
-        return 'bot-session';
-      },
-      async prepareSession() {
-        return 'ready';
-      },
+    sessions: createTestBotSessionAdapter({
       async runTurn(input) {
+        if (input.admissionMode === 'replay_only') return { kind: 'admission_required' };
         input.onReplySnapshot?.('Hello');
         input.onReplySnapshot?.('Hello world');
         return { kind: 'completed', text: 'Hello world' };
       },
-    },
+    }),
   });
 
   await service.handleBotIncomingMessage({
     platform: 'telegram',
     userId: 'user',
     userName: 'User',
-    chatId: 'chat',
+    conversationId: 'chat',
+    sourceEventId: 'source',
+    replyTarget: { chatId: 'chat', replyToMessageId: 'source' },
     isGroup: false,
     text: 'hello',
-    sourceMessageId: 'source',
     receivedAt: Date.now(),
   } as BotIncomingMessage);
 

apps/desktop/src/main/__tests__/bot-session-adapter-fixture.ts

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import type { BotSessionAdapter } from '../bot-session-adapter.js';
+
+export function createTestBotSessionAdapter(
+  overrides: Partial<BotSessionAdapter> = {},
+): BotSessionAdapter {
+  return {
+    async resolveSession() {
+      return { kind: 'ready', sessionId: 'bot-session-1' };
+    },
+    async releaseConversation() {
+      return false;
+    },
+    async runTurn(input) {
+      if (input.admissionMode === 'replay_only') return { kind: 'admission_required' };
+      return { kind: 'completed', text: 'ok' };
+    },
+    ...overrides,
+  };
+}