fix(agents): complete delegated work handoff and resume waiting agents

Keep each ACP attempt active until its own terminal result, isolate managed Hosts from coordinator scheduling, and route peer results back to waiting agents.

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

packages/cli/src/commands/serve.ts

@@ -891,6 +891,7 @@ export const serveCommand: CommandModule<unknown, ServeArgs> = {
     const { runQwenServe } = await import('../serve/run-qwen-serve.js');
     try {
       const serveOptions = {
+        agentHostWorker: Boolean(argv['agent-host-server']),
         port: argv.port,
         hostname: argv.hostname,
         token: argv.token,

packages/cli/src/serve/agent-host-client.ts

@@ -296,12 +296,23 @@ async function executeAssignment(
         undefined,
         { promptId, modelPrompt: modelPrompt(assignment) },
       );
-      const turn = await options.bridge.getSessionTurnStatus(
-        sessionId,
-        undefined,
-        promptId,
-      );
-      summary = turn?.resultText?.trim();
+      for (;;) {
+        const turn = await options.bridge.getSessionTurnStatus(
+          sessionId,
+          undefined,
+          promptId,
+        );
+        if (turn?.promptId === promptId) {
+          if (turn.state === 'error' || turn.state === 'cancelled') {
+            throw new Error(turn.error?.message ?? 'Managed Agent cancelled.');
+          }
+          if (turn.state === 'completed') {
+            summary = turn.resultText?.trim();
+            break;
+          }
+        }
+        await delay(250);
+      }
     }
   } finally {
     clearInterval(renew);

packages/cli/src/serve/server.ts

@@ -1430,8 +1430,9 @@ export function createServeApp(
   // `Config.isAgentCollaborationEnabled` so a daemon and the sessions it hosts
   // cannot disagree about whether the feature is on.
   agentCollaborationEnabled =
-    process.env['QWEN_CODE_ENABLE_AGENT_COLLABORATION'] === '1' ||
-    liveSettingsAtBoot?.experimental?.agentCollaboration === true;
+    !opts.agentHostWorker &&
+    (process.env['QWEN_CODE_ENABLE_AGENT_COLLABORATION'] === '1' ||
+      liveSettingsAtBoot?.experimental?.agentCollaboration === true);
 
   const liveConfigAtBoot = liveSettingsAtBoot
     ? readLiveVoiceConfiguration(liveSettingsAtBoot)
@@ -3166,7 +3167,7 @@ export function createServeApp(
         ? { deliverChannelMessage: deps.deliverChannelMessage }
         : {}),
     });
-  } else {
+  } else if (!opts.agentHostWorker) {
     // Close out runs the switch left mid-flight (architecture §6). Recovery
     // cannot tell "the daemon crashed" from "the operator turned this off"
     // — both look like a live run whose body is gone — so if these were left

packages/cli/src/serve/types.ts

@@ -340,6 +340,8 @@ export interface ServeOptions {
    * integer. Default: 10000 (10 s).
    */
   initializeTimeoutMs?: number;
+  /** A remote Host executes assignments; it does not own local scheduling. */
+  agentHostWorker?: boolean;
   /**
    * ACP session load/resume timeout in ms. Defaults to 60000 (60 s), raised
    * to an explicitly set initialize timeout when that value is larger. An

packages/cli/src/serve/workspace-agents/session-dispatch-port.ts

@@ -24,6 +24,7 @@ import {
   LOCAL_AGENT_RUNTIME_ID,
   SessionService,
 } from '@qwen-code/qwen-code-core';
+import { setTimeout as delay } from 'node:timers/promises';
 import type {
   AgentBodyState,
   AgentDispatchPort,
@@ -48,7 +49,9 @@ export type AgentSessionBridge = Pick<
   | 'cancelSession'
   | 'getSessionStatsStatus'
 > &
-  Partial<Pick<AcpSessionBridge, 'updateSessionMetadata'>>;
+  Partial<
+    Pick<AcpSessionBridge, 'updateSessionMetadata' | 'getSessionTurnStatus'>
+  >;
 
 export interface CreateSessionDispatchPortInput {
   bridge: AgentSessionBridge;
@@ -87,6 +90,37 @@ export function createSessionDispatchPort(
   const executions = new Map<string, AgentBodyState>();
   const sessions = new SessionService(workspaceCwd);
 
+  async function waitForTurn(
+    sessionId: string,
+    promptId: string,
+  ): Promise<void> {
+    const getSessionTurnStatus = bridge.getSessionTurnStatus;
+    if (!getSessionTurnStatus) return;
+    // ACP sendPrompt acknowledges admission; the turn terminal arrives on the
+    // bridge afterwards. Keep the dispatcher claim alive until that terminal
+    // is visible, otherwise a live body can be closed while calling a thread
+    // tool.
+    for (;;) {
+      const status = await getSessionTurnStatus(
+        sessionId,
+        undefined,
+        promptId,
+      );
+      if (
+        status?.promptId === promptId &&
+        (status.state === 'completed' ||
+          status.state === 'cancelled' ||
+          status.state === 'error')
+      ) {
+        if (status.state === 'error') {
+          throw new Error(status.error?.message ?? 'Agent turn failed.');
+        }
+        return;
+      }
+      await delay(250);
+    }
+  }
+
   /**
    * Sends one turn to an agent's session, saying which run it is a turn of.
    *
@@ -125,6 +159,7 @@ export function createSessionDispatchPort(
         },
       },
     );
+    await waitForTurn(sessionId, deliveryId);
   };
 
   return {
@@ -240,9 +275,8 @@ export function createSessionDispatchPort(
               attempt,
             };
             executions.set(sessionId, execution);
-            // sendPrompt resolves at turn completion, not queue acceptance.
-            // Keep dispatch free to start peers and service cancellation.
-            void send(sessionId, prompt, runId, context).then(
+            // Wait for this attempt's terminal while dispatch services peers.
+            void send(sessionId, prompt, `${runId}:${attempt}`, context).then(
               () => {
                 if (executions.get(sessionId) === execution) {
                   executions.delete(sessionId);

packages/core/src/agents/workspace-agents/run-lifecycle.ts

@@ -34,6 +34,8 @@ import {
 import { requireAgentRunContext, type AgentRunContext } from './run-context.js';
 import type { Thread, ThreadEvent, ThreadMessage } from './types.js';
 import { isThreadTerminal } from './types.js';
+import { postMessageInTransaction } from './thread-actions.js';
+import { mentionToken } from './mentions.js';
 
 /** How an agent says its run is done. `unclosed` is recorded, never chosen. */
 export type RunCloseRequest =
@@ -293,21 +295,47 @@ export async function closeRunInTransaction(
   let next = thread;
   let message: ThreadMessage | undefined;
   if (input.request.kind !== 'waiting') {
-    const appended = appendMessage(
-      next,
-      {
-        authorKind: 'agent',
-        from: context.agentId,
-        authorNameSnapshot: authorName,
-        sourceRunId: run.id,
-        triggerKind: `thread_${input.request.kind}`,
-        text:
-          input.request.kind === 'blocked'
-            ? input.request.question
-            : input.request.summary,
-      },
-      now,
+    const waiters = agents.filter(
+      (agent) =>
+        agent.id !== context.agentId &&
+        thread.runs.some(
+          (entry) =>
+            entry.agentId === agent.id &&
+            entry.closeKind === 'waiting' &&
+            entry.closeAcknowledgedAtSequence === undefined &&
+            (entry.status === 'completed' || entry.status === 'finishing'),
+        ),
     );
+    const text =
+      input.request.kind === 'blocked'
+        ? input.request.question
+        : input.request.summary;
+    const appended =
+      waiters.length > 0
+        ? await postMessageInTransaction(
+            transaction,
+            thread.id,
+            {
+              authorKind: 'agent',
+              from: context.agentId,
+              sourceRunId: run.id,
+              triggerKind: `thread_${input.request.kind}`,
+              text: `${waiters.map(mentionToken).join(' ')}\n\n${text}`,
+            },
+            { agents, now, threadOverride: next },
+          )
+        : appendMessage(
+            next,
+            {
+              authorKind: 'agent',
+              from: context.agentId,
+              authorNameSnapshot: authorName,
+              sourceRunId: run.id,
+              triggerKind: `thread_${input.request.kind}`,
+              text,
+            },
+            now,
+          );
     next = appended.thread;
     message = appended.message;
   }