fix(agents): preserve Host recovery and read-only execution

2026-10-21 22:39:45 +00:00 · 2026-10-01 18:47:14 +08:00 · 2026-10-01 18:47:14 +08:00 · 8d7faadd32
commit 8d7faadd32
parent 53b15769e8
14 changed files with 323 additions and 90 deletions
--- a/docs/design/2026-09-11-agent-project-host-entry.md
+++ b/docs/design/2026-09-11-agent-project-host-entry.md
@ -10,6 +10,8 @@ The sidebar Agent entry opens the roster, task board, and runtime list. Collabor

 Qwen Host execution streams ACP reply, thought, and tool activity for the active prompt. The daemon persists bounded latest snapshots and requires the current Host, lease, and attempt for every update. A final result replaces its live preview; failed or cancelled runs retain partial output. Messages received during a run are queued for a successor turn, and replaying a result does not create duplicate work.

+Restarted or taken-over executions receive a new attempt and lease so their spend is counted separately. After each attempt, the Host flushes and closes the live session; later turns resume its persisted history without retaining an active session slot.
+
 This increment exposes only Qwen Code on a managed Host. It neither probes for nor launches Codex or Claude. Those providers require a separate isolated profile that can confine reads to the selected workspace and exclude ambient user MCP servers, plugins, hooks, skills, and credentials.

 ## Isolation and connection
@ -26,4 +28,4 @@ Enrollment, heartbeat, pickup, lease renewal, progress, and result submission al

 Attachment lasts for the daemon process lifetime; this is not an OS service installer. A Host handles assignments sequentially. The feature does not synchronize files, open a relay, traverse NAT, or adopt an existing desktop session.

-A prior cross-machine Qwen acceptance run connected an x86_64 Linux Host to an arm64 macOS coordinator through an explicitly trusted tunnel, dispatched one read-only task, streamed progress, and returned a file that existed only on the remote checkout. That evidence predates the final isolation fixes; the final PR relies on review and remote CI for those fixes because local test and build runs were intentionally not repeated during closeout.
+Acceptance reports in [the PR review thread](https://github.com/QwenLM/qwen-code/pull/12582) identify the tested source head, environments, model type, and UI evidence. Cross-machine evidence from an earlier head does not validate later source changes.
--- a/docs/design/2026-09-11-agent-project-host-entry.zh-CN.md
+++ b/docs/design/2026-09-11-agent-project-host-entry.zh-CN.md
@ -10,6 +10,8 @@

 Qwen Host 只上报当前 prompt 的 ACP 回复、思考和工具活动。服务端保存有界的最新快照，每次更新都校验当前 Host、租约和 attempt。最终结果替换实时预览；失败或取消会保留部分输出。运行期间收到的消息进入后继轮次，重复提交结果不会重复创建任务。

+重启或接管执行时分配新的 attempt 和租约，分别记录每次执行的消耗。每次尝试结束后，Host 刷盘并关闭活跃会话；后续轮次恢复持久化历史，不持续占用活跃会话额度。
+
 本次增量只开放受管 Host 上的 Qwen Code，不探测或启动 Codex、Claude。其他 provider 需要独立的隔离配置，能够把读取限制在选中工作区，并排除用户环境中的 MCP、插件、Hooks、Skills 和凭据，完成后再单独接入。

 ## 隔离与连接
@ -26,4 +28,4 @@ Host 使用短期注册 token 接入，随后保存限定范围的凭据。token

 接入只持续到 daemon 进程结束，不安装操作系统服务。单个 Host 串行处理任务。此功能不负责文件同步、公共中继、NAT 穿透，也不接管已有桌面会话。

-此前的跨机器 Qwen 验收通过显式可信隧道，将 x86_64 Linux Host 连接到 arm64 macOS 协调端，完成了一次只读任务派发、进度流式上报，并返回了只存在于远程检出的文件。该证据早于最终隔离修复；收尾阶段按约束未重复本地 test/build，最终修复由代码审查和远端 CI 验证。
+[PR 评审线程](https://github.com/QwenLM/qwen-code/pull/12582)中的验收报告标明测试对应的源码 head、环境、模型类型和 UI 证据。较早 head 的跨机器证据不能作为后续源码变更的验证。
--- a/packages/cli/src/acp-integration/session/Session.test.ts
+++ b/packages/cli/src/acp-integration/session/Session.test.ts
@ -36828,74 +36828,92 @@ describe('Session', () => {
          expect(executeSpy).not.toHaveBeenCalled();
        });

-        it('runs the host guard with final params and denies before execution', async () => {
-          mockConfig.getApprovalMode = vi
-            .fn()
-            .mockReturnValue(ApprovalMode.YOLO);
-          const guard = vi.fn().mockResolvedValue({
-            allowed: false,
-            reason: 'host policy denied',
-          });
-          mockConfig.getToolInvocationGuard = vi.fn().mockReturnValue(guard);
+        it.each([false, true])(
+          'runs the host guard before execution (agent-host=%s)',
+          async (agentHost) => {
+            mockConfig.getApprovalMode = vi
+              .fn()
+              .mockReturnValue(
+                agentHost ? ApprovalMode.PLAN : ApprovalMode.YOLO,
+              );
+            mockConfig.getSessionSourceType = vi
+              .fn()
+              .mockReturnValue(agentHost ? 'agent-host' : undefined);
+            const guard = vi.fn().mockResolvedValue({
+              allowed: false,
+              reason: 'host policy denied',
+            });
+            mockConfig.getToolInvocationGuard = vi.fn().mockReturnValue(guard);

-          const executeSpy = vi.fn();
-          const tool = {
-            name: 'read_file',
-            kind: core.Kind.Read,
-            build: vi.fn().mockReturnValue({
-              params: { path: '/normalized/final.txt' },
-              getDefaultPermission: vi.fn().mockResolvedValue('allow'),
-              execute: executeSpy,
-            }),
-          };
+            const executeSpy = vi.fn();
+            const tool = {
+              name: 'read_file',
+              kind: core.Kind.Read,
+              build: vi.fn().mockReturnValue({
+                params: { path: '/normalized/final.txt' },
+                getDefaultPermission: vi
+                  .fn()
+                  .mockResolvedValue(agentHost ? 'ask' : 'allow'),
+                getConfirmationDetails: vi.fn().mockResolvedValue({
+                  type: 'exec',
+                  title: 'Confirm read',
+                  command: 'read',
+                  rootCommand: 'read',
+                  onConfirm: vi.fn(),
+                }),
+                execute: executeSpy,
+              }),
+            };

-          mockToolRegistry.getTool.mockReturnValue(tool);
-          mockChat.sendMessageStream = vi.fn().mockResolvedValue(
-            createStreamWithChunks([
-              {
-                type: core.StreamEventType.CHUNK,
-                value: {
-                  functionCalls: [
-                    {
-                      id: 'call-guard',
-                      name: 'read_file',
-                      args: { path: './original.txt' },
-                    },
-                  ],
+            mockToolRegistry.getTool.mockReturnValue(tool);
+            mockChat.sendMessageStream = vi.fn().mockResolvedValue(
+              createStreamWithChunks([
+                {
+                  type: core.StreamEventType.CHUNK,
+                  value: {
+                    functionCalls: [
+                      {
+                        id: 'call-guard',
+                        name: 'read_file',
+                        args: { path: './original.txt' },
+                      },
+                    ],
+                  },
                },
-              },
-            ]),
-          );
+              ]),
+            );

-          await session.prompt({
-            sessionId: 'test-session-id',
-            prompt: [{ type: 'text', text: 'read the file' }],
-          });
+            await session.prompt({
+              sessionId: 'test-session-id',
+              prompt: [{ type: 'text', text: 'read the file' }],
+            });

-          expect(guard).toHaveBeenCalledWith({
-            callId: 'call-guard',
-            toolName: 'read_file',
-            args: { path: '/normalized/final.txt' },
-            signal: expect.any(AbortSignal),
-            permissionChecked: true,
-            // The daemon policy falls back to the session and needs to know
-            // where the tool will run.
-            sessionId: 'test-session-id',
-            cwd: process.cwd(),
-          });
-          expect(executeSpy).not.toHaveBeenCalled();
-          expect(
-            mockChatRecordingService.recordToolResult,
-          ).toHaveBeenCalledWith(
-            expect.any(Array),
-            expect.objectContaining({
+            expect(guard).toHaveBeenCalledWith({
              callId: 'call-guard',
-              status: 'error',
-              executionStatus: 'not_started',
-              errorType: core.ToolErrorType.EXECUTION_DENIED,
-            }),
-          );
-        });
+              toolName: 'read_file',
+              args: { path: '/normalized/final.txt' },
+              signal: expect.any(AbortSignal),
+              permissionChecked: !agentHost,
+              // The daemon policy falls back to the session and needs to know
+              // where the tool will run.
+              sessionId: 'test-session-id',
+              cwd: process.cwd(),
+            });
+            expect(executeSpy).not.toHaveBeenCalled();
+            expect(mockClient.requestPermission).not.toHaveBeenCalled();
+            expect(
+              mockChatRecordingService.recordToolResult,
+            ).toHaveBeenCalledWith(
+              expect.any(Array),
+              expect.objectContaining({
+                callId: 'call-guard',
+                status: 'error',
+                executionStatus: 'not_started',
+                errorType: core.ToolErrorType.EXECUTION_DENIED,
+              }),
+            );
+          },
+        );

        it('executes once when the host guard allows the final invocation', async () => {
          mockConfig.getApprovalMode = vi
--- a/packages/cli/src/acp-integration/session/Session.ts
+++ b/packages/cli/src/acp-integration/session/Session.ts
@ -13972,6 +13972,36 @@ export class Session implements SessionContext {
          callIdAware.setCallId?.(callId);
          toolBuildSucceeded = true;

+          if (this.config.getSessionSourceType?.() === 'agent-host') {
+            const guard = this.config.getToolInvocationGuard?.();
+            if (guard) {
+              const invocationContext = getInvocationContext();
+              const decision = await evaluateToolInvocationGuard(guard, {
+                callId,
+                toolName: policyToolName,
+                args: invocation.params as Record<string, unknown>,
+                signal: activeToolAbortSignal,
+                permissionChecked: false,
+                sessionId: this.config.getSessionId(),
+                cwd: this.config.getTargetDir(),
+                ...(invocationContext ? { invocationContext } : {}),
+              });
+              const cancellation = cancelBeforeExecutionIfAborted(toolName);
+              if (cancellation) return cancellation;
+              if (!decision.allowed) {
+                return earlyErrorResponse(
+                  new Error(decision.reason),
+                  toolName,
+                  {
+                    status: 'error',
+                    errorType: ToolErrorType.EXECUTION_DENIED,
+                    executionStatus: 'not_started',
+                  },
+                );
+              }
+            }
+          }
+
          // Production AgentTool always initializes `eventEmitter` on its
          // invocation (`agent.ts:392`). Be defensive about the `undefined`
          // case too so an incomplete/custom AgentTool invocation degrades
--- a/packages/cli/src/config/config.test.ts
+++ b/packages/cli/src/config/config.test.ts
@ -5199,6 +5199,7 @@ describe('loadCliConfig with includeDirectories', () => {
      expect.objectContaining({
        targetDir: mockCwd,
        safeMode: true,
+        coreTools: [ToolNames.READ_FILE, ToolNames.GREP, ToolNames.LS],
        includeDirectories: [],
        lsp: { enabled: false },
        disableAllHooks: true,
--- a/packages/cli/src/config/config.ts
+++ b/packages/cli/src/config/config.ts
@ -2873,6 +2873,11 @@ export async function loadCliConfig(
  };

  if (agentHostReadOnly) {
+    configParams.coreTools = [
+      ToolNames.READ_FILE,
+      ToolNames.GREP,
+      ToolNames.LS,
+    ];
    configParams.disableAllHooks = true;
    configParams.mcpServers = {};
    configParams.topTierMcpServers = undefined;
--- a/packages/cli/src/serve/agent-host-client.test.ts
+++ b/packages/cli/src/serve/agent-host-client.test.ts
@ -4,8 +4,13 @@
 * SPDX-License-Identifier: Apache-2.0
 */

-import { describe, expect, it } from 'vitest';
-import { isRevocation } from './agent-host-client.js';
+import { createHash } from 'node:crypto';
+import * as fs from 'node:fs/promises';
+import os from 'node:os';
+import path from 'node:path';
+import { describe, expect, it, vi } from 'vitest';
+import type { AcpSessionBridge } from './acp-session-bridge.js';
+import { isRevocation, startAgentHostConnection } from './agent-host-client.js';

 describe('isRevocation', () => {
  // The transport error carries the route's body text as its message.
@ -38,3 +43,45 @@ describe('isRevocation', () => {
    expect(isRevocation(undefined)).toBe(false);
  });
 });
+
+it('retains a saved credential when rejoining sees only bare bearer 401s', async () => {
+  const qwenDir = await fs.mkdtemp(path.join(os.tmpdir(), 'pr12582-f1-'));
+  const serverUrl = 'http://127.0.0.1:18582';
+  const workspaceId = 'ws-test';
+  const workspaceCwd = '/pr12582-test';
+  const key = createHash('sha256')
+    .update(`${serverUrl}\0${workspaceId}\0${workspaceCwd}`)
+    .digest('hex');
+  const file = path.join(qwenDir, 'agent-hosts', `${key}.json`);
+  const credential = JSON.stringify({
+    schemaVersion: 1,
+    serverUrl,
+    workspaceId,
+    hostId: 'host-saved',
+    secret: 'saved-test-secret',
+  });
+  await fs.mkdir(path.dirname(file), { recursive: true });
+  await fs.writeFile(file, credential);
+  const fetchMock = vi.fn(async () =>
+    Response.json({ error: 'Unauthorized' }, { status: 401 }),
+  );
+  vi.stubEnv('QWEN_HOME', qwenDir);
+  vi.stubGlobal('fetch', fetchMock);
+  try {
+    await expect(
+      startAgentHostConnection({
+        bridge: {} as AcpSessionBridge,
+        serverUrl,
+        workspaceId,
+        workspaceCwd,
+        enrollmentToken: 'the-original-ui-join-token',
+      }),
+    ).rejects.toThrow();
+    expect(fetchMock).toHaveBeenCalled();
+    await expect(fs.readFile(file, 'utf8')).resolves.toBe(credential);
+  } finally {
+    vi.unstubAllGlobals();
+    vi.unstubAllEnvs();
+    await fs.rm(qwenDir, { recursive: true, force: true });
+  }
+});
--- a/packages/cli/src/serve/agent-host-client.ts
+++ b/packages/cli/src/serve/agent-host-client.ts
@ -386,12 +386,14 @@ async function executeAssignment(
  progressHeartbeat.unref?.();
  renew.unref?.();
  let summary: string | undefined;
+  let tokens: number | undefined;
+  const sessionId = agentThreadSessionId(
+    `${credential.hostId}:${assignment.agent.id}`,
+    assignment.threadId,
+  );
+  let sessionReady = false;
  try {
    void flush();
-    const sessionId = agentThreadSessionId(
-      `${credential.hostId}:${assignment.agent.id}`,
-      assignment.threadId,
-    );
    const sourceId = `${credential.hostId}:${assignment.agent.id}`;
    const sessions = new SessionService(options.workspaceCwd);
    const live = options.bridge
@ -414,6 +416,7 @@ async function executeAssignment(
        });
      }
    }
+    sessionReady = true;
    stream = streamAgentTurn(
      options.bridge,
      sessionId,
@ -516,11 +519,23 @@ async function executeAssignment(
    updates.abort();
    await stream;
    await flush();
+    tokens = await measureTokens?.();
+    if (sessionReady) {
+      await options.bridge
+        .closeSession(sessionId, undefined, {
+          requireAgentClose: true,
+          agentCloseTimeoutMs: 10_000,
+        })
+        .catch((error: unknown) => {
+          writeStderrLine(
+            `Agent Host could not close session: ${String(error)}`,
+          );
+        });
+    }
  }
  if (!summary) {
    throw new Error('Managed Agent finished without a final answer.');
  }
-  const tokens = await measureTokens?.();
  return {
    threadId: assignment.threadId,
    runId: assignment.runId,
@ -705,7 +720,7 @@ async function connectAgentHost(
        await sendHeartbeat(savedCredential);
        throw new Error('Invalid or expired Agent Host enrollment token.');
      } catch (credentialError) {
-        if ((credentialError as { status?: number }).status !== 401) {
+        if (!isRevocation(credentialError)) {
          throw credentialError;
        }
        await fs.rm(filePath, { force: true });
--- a/packages/core/src/agents/workspace-agents/host-lease.test.ts
+++ b/packages/core/src/agents/workspace-agents/host-lease.test.ts
@ -126,7 +126,8 @@ describe('pickupRunForHost', () => {
    expect(followup.outcomes[0]?.decision.kind).toBe('coalesce');

    const replay = (await pickupRunForHost(PROJECT_ROOT, mine, T0 + 1))!;
-    expect(replay.lease.leaseId).toBe(first.lease.leaseId);
+    expect(replay.lease.leaseId).not.toBe(first.lease.leaseId);
+    expect(replay.attempt).toBe(first.attempt + 1);
    expect(replay.prompt).toContain('Also inspect the follow-up.');
    expect(
      await applyHostRunResult(
@ -204,6 +205,74 @@ describe('pickupRunForHost', () => {
 });

 describe('leases', () => {
+  it.each(['same host restart', 'expired lease takeover'])(
+    'charges each execution after %s and rejects the old result',
+    async (replay) => {
+      const firstHost = await host('first', ['qwen']);
+      const secondHost = await host('second', ['qwen']);
+      await placeAgent([firstHost, secondHost]);
+      const threadId = await seedQueued();
+      const startedAt = Date.now();
+      const first = (await pickupRunForHost(
+        PROJECT_ROOT,
+        firstHost,
+        startedAt,
+      ))!;
+      const firstIdentity = {
+        threadId,
+        runId: first.runId,
+        hostId: firstHost,
+        leaseId: first.lease.leaseId,
+        attempt: first.attempt,
+      };
+      await expect(
+        reportHostRunProgress(PROJECT_ROOT, {
+          ...firstIdentity,
+          sequence: 1,
+          stage: 'thinking',
+          detail: '',
+          tokens: 1_050,
+        }),
+      ).resolves.toEqual({ ok: true });
+      const at =
+        startedAt +
+        (replay === 'same host restart' ? 1 : DEFAULT_RUN_LEASE_MS + 1);
+      const next = (await pickupRunForHost(
+        PROJECT_ROOT,
+        replay === 'same host restart' ? firstHost : secondHost,
+        at,
+      ))!;
+      expect(next.attempt).toBe(first.attempt + 1);
+      const result = {
+        status: 'completed' as const,
+        close: { kind: 'review' as const, summary: 'Marker read.' },
+        tokens: 1_050,
+      };
+      await expect(
+        applyHostRunResult(PROJECT_ROOT, { ...firstIdentity, ...result }, at),
+      ).resolves.toMatchObject({ ok: false });
+      await expect(
+        applyHostRunResult(
+          PROJECT_ROOT,
+          {
+            ...firstIdentity,
+            hostId: next.lease.hostId,
+            leaseId: next.lease.leaseId,
+            attempt: next.attempt,
+            ...result,
+          },
+          at,
+        ),
+      ).resolves.toMatchObject({ ok: true });
+      const thread = (await readThread(PROJECT_ROOT, threadId))!;
+      const run = thread.runs[0];
+      expect(run.usageByRound.map((usage) => usage.tokens)).toEqual([
+        1_050, 1_050,
+      ]);
+      expect(thread.tokensUsed).toBe(2_100);
+    },
+  );
+
  it('will not revive an expired lease', async () => {
    const mine = await host('mine', ['qwen']);
    await placeAgent([mine]);
@ -242,7 +311,7 @@ describe('leases', () => {
        },
        later + 1,
      ),
-    ).resolves.toEqual({ ok: false, reason: 'stale_lease' });
+    ).resolves.toEqual({ ok: false, reason: 'attempt_moved_on' });
  });

  it('requeues a follow-up that a failed Host turn never received', async () => {
--- a/packages/core/src/agents/workspace-agents/host-lease.ts
+++ b/packages/core/src/agents/workspace-agents/host-lease.ts
@ -519,11 +519,12 @@ export async function pickupRunForHost(
          run.lease.expiresAt > now,
      );
    if (held?.agent && held.run.lease) {
+      const attempt = held.run.attempts + 1;
      const assignment = assignmentFor(
        transaction,
        held.agent,
        held.thread,
-        held.run,
+        { ...held.run, attempts: attempt },
        roster,
      );
      const committed =
@ -538,12 +539,15 @@ export async function pickupRunForHost(
        .map((message) => message.id);
      const lease = {
        ...held.run.lease,
+        leaseId: randomBytes(16).toString('hex'),
+        attempt,
        expiresAt: now + DEFAULT_RUN_LEASE_MS,
      };
      await transaction.writeThread(
        // The host restarted the turn, so its progress restarts too.
        withRun(held.thread, held.run.id, (run) => ({
          ...run,
+          attempts: attempt,
          lease,
          progress: undefined,
          acceptedMessageIds: Array.from(
@ -622,7 +626,7 @@ export async function pickupRunForHost(
            attempts: candidate.run.attempts + 1,
            startedAt: now,
          }
-        : candidate.run;
+        : { ...candidate.run, attempts: candidate.run.attempts + 1 };
    const lease: RunLease = {
      hostId,
      leaseId: randomBytes(16).toString('hex'),
--- a/packages/core/src/tools/tool-registry.test.ts
+++ b/packages/core/src/tools/tool-registry.test.ts
@ -187,6 +187,34 @@ describe('ToolRegistry', () => {
  let toolRegistry: ToolRegistry;
  let mockConfigGetToolDiscoveryCommand: ReturnType<typeof vi.spyOn>;

+  it('registers only runnable read tools for agent-host sessions through every registration path', async () => {
+    const config = new Config({
+      ...baseConfigParams,
+      safeMode: true,
+      coreTools: ['read_file', 'grep_search', 'list_directory'],
+    });
+    config.setSessionSource('agent-host', 'host_1');
+    const registry = await config.createToolRegistry(undefined, {
+      skipDiscovery: true,
+    });
+    for (const name of [
+      'agent',
+      'tool_call',
+      'mcp__ambient__read',
+      'unknown',
+    ]) {
+      const tool = new MockTool({ name });
+      registry.registerTool(tool);
+      registry.registerFactory(name, async () => tool);
+      registry.registerPermissionDeferredFactory(name, async () => tool);
+    }
+    expect(registry.getAllToolNames().sort()).toEqual([
+      'grep_search',
+      'list_directory',
+      'read_file',
+    ]);
+  });
+
  beforeEach(() => {
    vi.mocked(fs.existsSync).mockReturnValue(true);
    vi.mocked(fs.statSync).mockReturnValue({
--- a/packages/core/src/tools/tool-registry.ts
+++ b/packages/core/src/tools/tool-registry.ts
@ -312,8 +312,8 @@ export class ToolRegistry {
  }

  /**
-   * Returns true when `name` is in the Config's `disabledTools` set, in
-   * which case `registerTool` / `registerFactory` will skip it. This is
+   * Returns true when Config disables `name` or the Host profile withholds it,
+   * in which case tool registration will skip it. This is
   * the chokepoint for the daemon mutation route at `POST /workspace/
   * tools/:name/enable {enabled:false}`; both
   * built-ins and MCP-discovered tools flow through `registerTool`, so
@ -323,6 +323,14 @@ export class ToolRegistry {
    name: string,
    aliases: readonly string[] = [],
  ): boolean {
+    if (
+      this.config.getSessionSourceType?.() === 'agent-host' &&
+      name !== ToolNames.READ_FILE &&
+      name !== ToolNames.GREP &&
+      name !== ToolNames.LS
+    ) {
+      return true;
+    }
    const disabledTools = this.config.getDisabledTools();
    const hasExactMatch =
      disabledTools.has(name) ||
@ -351,7 +359,7 @@ export class ToolRegistry {
      )
    ) {
      debugLogger.info(
-        `Tool "${tool.name}" skipped: present in disabledTools set.`,
+        `Tool "${tool.name}" skipped: disabled for this session.`,
      );
      return;
    }
@ -416,7 +424,7 @@ export class ToolRegistry {
  registerFactory(name: string, factory: ToolFactory): void {
    if (this.isToolDisabled(name)) {
      debugLogger.info(
-        `Tool factory "${name}" skipped: present in disabledTools set.`,
+        `Tool factory "${name}" skipped: disabled for this session.`,
      );
      return;
    }
@ -439,7 +447,7 @@ export class ToolRegistry {
  registerPermissionDeferredFactory(name: string, factory: ToolFactory): void {
    if (this.isToolDisabled(name)) {
      debugLogger.info(
-        `Tool factory "${name}" skipped: present in disabledTools set.`,
+        `Tool factory "${name}" skipped: disabled for this session.`,
      );
      return;
    }
--- a/packages/web-shell/client/components/workspace-agents/ThreadChat.tsx
+++ b/packages/web-shell/client/components/workspace-agents/ThreadChat.tsx
@ -749,7 +749,9 @@ export function ThreadChat({
            !thread.posts.some(
              (post) =>
                post.sourceRunId === run.id &&
-                post.text.trim() === run.progress?.outputText?.trim(),
+                (run.closeKind === 'review' ||
+                  run.closeKind === 'blocked' ||
+                  post.text.trim() === run.progress?.outputText?.trim()),
            ),
        )
        .map(
--- a/packages/web-shell/client/e2e/web-shell.mesh-streaming.spec.ts
+++ b/packages/web-shell/client/e2e/web-shell.mesh-streaming.spec.ts
@ -280,23 +280,25 @@ test('mesh shows growing replies before completion, survives reload, and replace
      sourceRunId: run.id,
      authorKind: 'agent',
      authorName: agent.name,
-      text: 'Ready for review.',
+      text: 'Second fragment.',
      at: Date.now(),
    },
  ];
  await expect(
    transcript.getByText('First fragment. Second fragment.', { exact: true }),
-  ).toHaveCount(1);
+  ).toHaveCount(0);
  await expect(
-    transcript.getByText('Ready for review.', { exact: true }),
-  ).toBeVisible();
+    transcript.getByText('Second fragment.', { exact: true }),
+  ).toHaveCount(1);
  await expect(page.getByRole('button', { name: '验收并完成' })).toBeVisible();
  await page.reload();
-  const completedReply = transcript.getByText(
-    'First fragment. Second fragment.',
-    { exact: true },
-  );
+  const completedReply = transcript.getByText('Second fragment.', {
+    exact: true,
+  });
  await expect(completedReply).toBeVisible();
+  await expect(
+    transcript.getByText('First fragment. Second fragment.', { exact: true }),
+  ).toHaveCount(0);
  await page.screenshot({
    path: info.outputPath('02-completed.png'),
    animations: 'disabled',