patch

Author: tripolskypetr

TODO.md

@@ -387,7 +387,33 @@ shared state/storage и кэш эмбеддингов. Багов не найд
 - бан политики бьёт только по забаненному клиенту, сосед работает;
 - dispose одного клиента посреди медленного выполнения другого не ломает его.
 
-## Найденные и исправленные баги (21 итого)
+## 19-й заход: двойная отправка одной сессией без ожидания ответа (+5 тестов, +1 баг)
+
+### Новые тесты (5), сьют вырос до 223/223 — test/spec/doublesend.test.mjs
+- второй complete во время МЕДЛЕННОГО ТУЛА первого → busy-lock сериализует,
+  попарность сохранена;
+- fire-and-forget: три complete без ожидания предыдущих → каждый получает свой ответ;
+- makeConnection: два send без await → connector получает оба ответа по порядку;
+- cancelOutput посреди выполнения → первый complete возвращает "", следующий
+  получает СВОЙ ответ (регрессионный тест бага №22);
+- emit-подмена посреди выполнения → первый complete возвращает подменённый текст,
+  следующий получает СВОЙ ответ (регрессионный тест бага №22).
+
+### Баг №22: осиротевший вывод после cancelOutput/emit отравлял следующий обмен
+Файлы: `src/client/ClientAgent.ts`, `src/client/ClientSwarm.ts` (найден D4/D5)
+- cancelOutput (или emit-подмена) резолвили ожидателя текущего complete, но само
+  выполнение продолжало жить и ПОЗЖЕ эмитило свой результат в _outputSubject.
+  К этому моменту уже был подписан ожидатель СЛЕДУЮЩЕГО сообщения — пользователь
+  получал ответ на отменённый вопрос (r2="STALE-ANSWER" вместо echo).
+- Исправление: эпоха вывода _outputEpoch в ClientAgent. EXECUTE_FN захватывает
+  эпоху на старте и передаёт её во все _emitOutput (включая late-recovery в
+  createToolCall); commitCancelOutput инкрементит эпоху;
+  ClientAgent.commitOutputSubstituted() — то же для emit (вызывается из
+  ClientSwarm.emit по всем агентам). _emitOutput с несовпавшей эпохой молча
+  дропает stale-результат (двойная проверка: на входе и прямо перед
+  _outputSubject.next, т.к. между ними есть await'ы).
+
+## Найденные и исправленные баги (22 итого)
 
 ### 1. Дедлок waitForOutput при functools-kit v4 (причина 39 упавших тестов)
 Файл: `src/client/ClientSwarm.ts`

src/client/ClientAgent.ts

@@ -110,6 +110,7 @@ const createToolCall = async (
   targetFn: IAgentTool,
   reason: string,
   mode: ExecutionMode,
+  outputEpoch: number,
   self: ClientAgent
 ) => {
   self._runningToolCalls += 1;
@@ -192,7 +193,7 @@ const createToolCall = async (
           tool.function.name
         } error=${getErrorMessage(error)}`
       );
-      await self._emitOutput(mode, result);
+      await self._emitOutput(mode, result, outputEpoch);
     }
   } finally {
     self._runningToolCalls -= 1;
@@ -359,6 +360,7 @@ const EXECUTE_FN = async (
       incoming,
       mode
     );
+  const outputEpoch = self._outputEpoch;
   await self.params.history.push({
     role: "user",
     mode,
@@ -431,7 +433,7 @@ const EXECUTE_FN = async (
           self.params.logger.debug(
             `ClientAgent agentName=${self.params.agentName} clientId=${self.params.clientId} execute end result=${result}`
           );
-        await self._emitOutput(mode, result);
+        await self._emitOutput(mode, result, outputEpoch);
         run(false);
         return;
       }
@@ -485,7 +487,7 @@ const EXECUTE_FN = async (
           self.params.logger.debug(
             `ClientAgent agentName=${self.params.agentName} clientId=${self.params.clientId} execute end result=${result}`
           );
-        await self._emitOutput(mode, result);
+        await self._emitOutput(mode, result, outputEpoch);
         run(false);
         return;
       }
@@ -556,6 +558,7 @@ const EXECUTE_FN = async (
           targetFn,
           message.content || "",
           mode,
+          outputEpoch,
           self
         );
         const status = await statusAwaiter;
@@ -626,7 +629,7 @@ const EXECUTE_FN = async (
             self.params.logger.debug(
               `ClientAgent agentName=${self.params.agentName} clientId=${self.params.clientId} execute end result=${result}`
             );
-          await self._emitOutput(mode, result);
+          await self._emitOutput(mode, result, outputEpoch);
         }
         return status;
       });
@@ -669,14 +672,14 @@ const EXECUTE_FN = async (
       mode,
       `Invalid model output: ${result}`
     );
-    await self._emitOutput(mode, result1);
+    await self._emitOutput(mode, result1, outputEpoch);
     return;
   }
   GLOBAL_CONFIG.CC_LOGGER_ENABLE_DEBUG &&
     self.params.logger.debug(
       `ClientAgent agentName=${self.params.agentName} clientId=${self.params.clientId} execute end result=${result}`
     );
-  await self._emitOutput(mode, result);
+  await self._emitOutput(mode, result, outputEpoch);
 };
 
 /**
@@ -710,6 +713,15 @@ export class ClientAgent implements IAgent {
    */
   _activeToolChains = 0;
 
+  /**
+   * Generation counter for output emissions. Bumped by commitCancelOutput and by
+   * swarm-level output substitution (emit). Executions capture the epoch at start
+   * and _emitOutput drops their result when the epoch moved on — otherwise the
+   * stale output of a cancelled/substituted execution would resolve the waiter
+   * of the NEXT message, poisoning that exchange.
+   */
+  _outputEpoch = 0;
+
   /**
    * Subject for signaling agent changes, halting subsequent tool executions via commitAgentChange.
    * @readonly
@@ -915,7 +927,19 @@ export class ClientAgent implements IAgent {
    * Supports SwarmConnectionService by broadcasting agent outputs within the swarm.
    * @throws {Error} If validation fails after model resurrection, indicating an unrecoverable state.
    **/
-  async _emitOutput(mode: ExecutionMode, rawResult: string): Promise<void> {
+  async _emitOutput(
+    mode: ExecutionMode,
+    rawResult: string,
+    outputEpoch = this._outputEpoch
+  ): Promise<void> {
+    if (outputEpoch !== this._outputEpoch) {
+      GLOBAL_CONFIG.CC_LOGGER_ENABLE_DEBUG &&
+        this.params.logger.debug(
+          `ClientAgent agentName=${this.params.agentName} clientId=${this.params.clientId} _emitOutput skipped for stale execution (cancelled or substituted)`,
+          { mode, rawResult }
+        );
+      return;
+    }
     const result = await this.params.transform(
       rawResult,
       this.params.clientId,
@@ -939,6 +963,9 @@ export class ClientAgent implements IAgent {
           `agent-swarm-kit ClientAgent agentName=${this.params.agentName} clientId=${this.params.clientId} model resurrect failed: ${validation}`
         );
       }
+      if (outputEpoch !== this._outputEpoch) {
+        return;
+      }
       this.params.onOutput &&
         this.params.onOutput(
           this.params.clientId,
@@ -969,6 +996,9 @@ export class ClientAgent implements IAgent {
       });
       return;
     }
+    if (outputEpoch !== this._outputEpoch) {
+      return;
+    }
     this.params.onOutput &&
       this.params.onOutput(this.params.clientId, this.params.agentName, result);
     await this._outputSubject.next(result);
@@ -1094,6 +1124,19 @@ export class ClientAgent implements IAgent {
     return placeholder;
   }
 
+  /**
+   * Marks in-flight executions as substituted: the swarm emitted a replacement
+   * output (ClientSwarm.emit), so results of executions started earlier must not
+   * reach _outputSubject — they would pair with the next waiter otherwise.
+   */
+  commitOutputSubstituted(): void {
+    GLOBAL_CONFIG.CC_LOGGER_ENABLE_DEBUG &&
+      this.params.logger.debug(
+        `ClientAgent agentName=${this.params.agentName} clientId=${this.params.clientId} commitOutputSubstituted`
+      );
+    this._outputEpoch += 1;
+  }
+
   /**
    * Waits for the next output to be emitted via _outputSubject, typically after execute or run.
    * Useful for external consumers (e.g., SwarmConnectionService) awaiting agent responses.
@@ -1366,6 +1409,7 @@ export class ClientAgent implements IAgent {
       this.params.logger.debug(
         `ClientAgent agentName=${this.params.agentName} clientId=${this.params.clientId} commitCancelOutput`
       );
+    this._outputEpoch += 1;
     this._toolAbortController.abort();
     await this._cancelOutputSubject.next(CANCEL_OUTPUT_SYMBOL);
     await this.params.bus.emit<IBusEvent>(this.params.clientId, {

src/client/ClientSwarm.ts

@@ -367,6 +367,15 @@ export class ClientSwarm implements ISwarm {
       this.params.logger.debug(
         `ClientSwarm swarmName=${this.params.swarmName} clientId=${this.params.clientId} emit`
       );
+    // The emitted message substitutes the output of any in-flight execution:
+    // invalidate agents' pending emissions so a stale result cannot resolve
+    // the waiter of the next message.
+    for (const [, agent] of this._agentList) {
+      const agentRef = agent as unknown as {
+        commitOutputSubstituted?: () => void;
+      };
+      agentRef.commitOutputSubstituted?.();
+    }
     await this._emitSubject.next({
       agentName: await this.getAgentName(),
       output: message,

test/index.mjs

@@ -31,6 +31,7 @@ import "./spec/loop.test.mjs";
 import "./spec/toolguard.test.mjs";
 import "./spec/aliastools.test.mjs";
 import "./spec/multiclient.test.mjs";
+import "./spec/doublesend.test.mjs";
 
 run(import.meta.url, () => {
     console.log("All tests are finished");