---
title: "fix: WebSocket 断开后任务结果丢失与断线恢复"
status: completed
created: 2026-06-17
type: fix
origin: in-session investigation
---

# fix: WebSocket 断开后任务结果丢失与断线恢复

## Summary

当用户在复杂任务执行过程中刷新页面时，WebSocket 断开导致 ReAct 任务中断、已收集的输出丢失、且无法恢复。本计划通过三层防御彻底解决：Layer 1 确保部分结果持久化，Layer 2 将任务后台化解耦 WebSocket 生命周期，Layer 3 前端断线后恢复进行中任务。

## Problem Frame

**症状**：用户布置复杂任务后刷新界面，操作停止且返回为空。

**根因**（经代码调查确认）：

1. **结果丢失**：portal.py 的 ReAct 流式路径中，assistant 回复保存在 `async for` 循环之后（portal.py:1064），WebSocket 断开时该行永远不执行，`collected_output` 全部丢失。
2. **任务中断**：ReAct 任务通过 `async for event in react_engine.execute_stream(...)` 直接在 WebSocket 协程中执行，未使用 BackgroundRunner 后台化，WebSocket 断开 = 任务取消。
3. **无状态追踪**：portal WebSocket 路径不使用 TaskStore，`task_id` 仅用于 EventQueue 事件发射，任务状态无法查询。
4. **无恢复机制**：前端 WebSocket 重连后（3 秒自动重连），不检查是否有未完成任务，不恢复进行中的任务状态。

## Requirements

- **R1**: WebSocket 断开时，已收集的 ReAct 输出必须持久化到 conversation store
- **R2**: ReAct 任务必须后台执行，与 WebSocket 连接生命周期解耦
- **R3**: 每个任务必须在 TaskStore 中注册，状态可查询
- **R4**: 后台任务的事件必须通过 EventQueue 分发，WebSocket 可订阅
- **R5**: 前端 WebSocket 重连后，必须检查当前对话是否有未完成任务并恢复
- **R6**: 已完成的任务结果必须能从 conversation store 恢复显示
- **R7**: 不破坏现有 DIRECT_CHAT 路径和 REST API 路径

## Key Technical Decisions

### KTD1: 复用 EventQueue 作为任务事件总线

**决策**：使用现有的 `EventQueue`（`core/event_queue.py`）分发后台任务事件，而非新建 EventBus 模块。

**理由**：EventQueue 已具备所需能力——多订阅者广播（行 161-175）、缓冲回放（deque 100，行 153）、原子订阅（行 193-202）、哨兵关闭模式（行 230-243）。它已被 portal.py 的 `_emit_event_safe` 使用，且文件头注释明确说明设计目标包括 Portal WebSocket 订阅。

### KTD2: 任务后台化使用 asyncio.create_task + EventQueue 订阅

**决策**：在 portal.py 中，将 ReAct 执行包装为 `asyncio.create_task`，WebSocket 协程通过 `event_queue.subscribe()` 订阅事件流。任务在后台独立运行，WebSocket 仅转发事件给前端。

**理由**：BackgroundRunner（`server/runner.py`）虽然存在，但它调用 `agent.execute()`（非流式），不支持事件流。改造 BackgroundRunner 支持流式会侵入核心执行路径。直接在 portal.py 中用 `asyncio.create_task` 更聚焦、风险更低。

### KTD3: 任务状态通过 TaskStore 追踪

**决策**：在 portal.py WebSocket 路径中，为每个用户消息创建 TaskStore 记录，状态随任务进展更新（PENDING → RUNNING → COMPLETED/FAILED）。

**理由**：TaskStore 已在 `app.state.task_store` 可用，`/api/v1/tasks/{task_id}` 端点已存在。前端可通过现有 REST API 查询任务状态，无需新建端点。

### KTD4: 前端通过 conversation_id 关联任务

**决策**：在 TaskStore 记录的 `metadata` 中存储 `conversation_id`，前端重连后通过 `GET /api/v1/tasks?status=running` 查找当前对话的未完成任务。

**理由**：避免新建专门的"按 conversation 查任务"端点。前端遍历 running 任务，匹配 metadata 中的 conversation_id 即可。

## High-Level Technical Design

```
┌──────────────────────────────────────────────────────────────────┐
│                     WebSocket 连接存活                            │
│  ┌────────────┐    subscribe(task_id)    ┌──────────────────┐   │
│  │  WebSocket │ ◄────────────────────── │   EventQueue     │   │
│  │  协程      │    转发事件给前端         │  (缓冲回放+广播)  │   │
│  └────────────┘                          └────────┬─────────┘   │
│                                                  │ emit         │
├──────────────────────────────────────────────────┼─────────────┤
│                     WebSocket 断开                │             │
│  ┌────────────┐                                 │             │
│  │  WebSocket │ ✗ 断开                          │             │
│  │  协程      │                                 │             │
│  └────────────┘                                 │             │
│                                                  ▼             │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │            后台任务 (asyncio.create_task)                │   │
│  │  ┌─────────────┐  ┌──────────────┐  ┌────────────────┐ │   │
│  │  │ ReAct 执行  │─►│ EventQueue   │  │ TaskStore      │ │   │
│  │  │ execute_    │  │ .emit(event) │  │ .update_status │ │   │
│  │  │ stream()    │  │              │  │                │ │   │
│  │  └─────────────┘  └──────────────┘  └────────────────┘ │   │
│  │         │                                            │   │
│  │         ▼                                            │   │
│  │  ┌────────────────────────────────────────────────┐  │   │
│  │  │ conversation_store.add_message(assistant, ...)  │  │   │
│  │  │ (无论 WebSocket 是否存活，结果都持久化)          │  │   │
│  │  └────────────────────────────────────────────────┘  │   │
│  └─────────────────────────────────────────────────────────┘   │
│                                                                │
│  前端重连后:                                                   │
│  1. GET /api/v1/tasks?status=running → 查找未完成任务          │
│  2. 匹配 metadata.conversation_id → 当前对话的任务              │
│  3. event_queue.subscribe(task_id) → 恢复事件流                │
│  4. 或 GET /conversations/{id} → 拉取已完成的结果              │
└──────────────────────────────────────────────────────────────────┘
```

## Scope Boundaries

### In Scope

- portal.py WebSocket 路径的 ReAct 任务后台化
- WebSocket 断开时的部分结果持久化
- TaskStore 注册和状态更新
- EventQueue 事件分发集成
- 前端 WebSocket 重连后的任务恢复
- 前端任务状态查询 API 客户端

### Out of Scope

- BackgroundRunner 改造支持流式（风险过高，单独处理）
- REST API 路径（`/portal/chat`）的改造（REST 已是同步调用，无此问题）
- SSE 路径（`/portal/chat/stream`）的改造
- Expert Team 协作模式的后台化（单独处理）
- WebSocket 重连退避策略优化（独立改进）

### Deferred to Follow-Up Work

- SubmissionQueue 接入（目前完全闲置，可后续用于任务队列化）
- Redis 分布式任务恢复（当前 InMemoryTaskStore 足够，分布式场景后续处理）
- 任务进度百分比反馈（当前 ReAct 无细粒度进度概念）

---

## Implementation Units

### U1. Layer 1: WebSocket 断开时持久化已收集的输出

**Goal**: 确保 WebSocket 在 ReAct 流式过程中断开时，已收集的 `collected_output` 保存到 conversation store。

**Requirements**: R1

**Dependencies**: 无

**Files**:
- `src/agentkit/server/routes/portal.py` — 修改 ReAct 流式路径的异常处理
- `tests/unit/test_portal_ws_persistence.py` — 新建测试

**Approach**:

在 portal.py 的 `portal_websocket` 函数中，ReAct 流式路径（约 1014-1064 行）的异常处理需要增强：

1. 在 `except Exception as e:` 块（约 1049 行）中，在发送 error 之前，将 `collected_output` 保存到 `_conversation_store`。使用 `_ensure_non_empty` 处理空输出。
2. 在外层 `except WebSocketDisconnect:` 块（约 1102 行）中，检查是否有未保存的 `collected_output`，如果有则保存。
3. 在外层 `except Exception as e:` 块（约 1104 行）中，同样处理。
4. 需要将 `collected_output`、`conv`、`task_id` 声明提升到外层 try 作用域，使 except 块可以访问。

**Patterns to follow**:
- `_ensure_non_empty()` 函数（portal.py:58-62）处理空输出
- `_conversation_store.add_message()` 已有调用模式（portal.py:871, 952, 1064）
- `_emit_event_safe()` 的异常吞咽模式（portal.py:65-95）——保存操作失败不应阻断后续清理

**Test scenarios**:
- **Happy path**: ReAct 正常完成，`collected_output` 有内容，WebSocket 未断开 → 结果保存到 conversation store（验证现有行为不被破坏）
- **WebSocket 断开（final_answer 前）**: ReAct 流式中 WebSocket 断开，`collected_output` 为空 → conversation store 不写入空消息（或写入 EMPTY_LLM_RESPONSE）
- **WebSocket 断开（final_answer 后）**: ReAct 已产出 final_answer，WebSocket 在后续步骤断开，`collected_output` 有部分内容 → 部分内容保存到 conversation store
- **异常路径**: ReAct 执行抛出异常，`collected_output` 有部分内容 → 部分内容保存，error 事件正常发射
- **Integration**: 保存操作自身失败（conversation_store 异常）→ 不阻断 error 事件发射和后续清理

**Verification**: 模拟 WebSocket 断开场景，检查 SQLite conversation store 中是否有 assistant 消息记录。

---

### U2. Layer 2: ReAct 任务后台化与 EventQueue 事件分发

**Goal**: 将 ReAct 执行从 WebSocket 协程中解耦，后台执行并通过 EventQueue 分发事件。WebSocket 仅订阅事件流并转发给前端。

**Requirements**: R2, R4

**Dependencies**: U1

**Files**:
- `src/agentkit/server/routes/portal.py` — 重构 ReAct 流式路径
- `src/agentkit/core/event_queue.py` — 可能需要扩展（添加 task_id 过滤订阅）
- `tests/unit/test_portal_ws_background_task.py` — 新建测试

**Approach**:

1. **提取后台执行函数**：创建 `_execute_react_background()` 协程，接收所有必要参数（messages, tools, model, agent_name, system_prompt, timeout_seconds, conv_id, task_id, event_queue, conversation_store）。该函数：
   - 执行 `react_engine.execute_stream()`
   - 每个事件通过 `event_queue.emit()` 分发
   - `final_answer` 事件时累积输出
   - 正常结束后保存到 `conversation_store`
   - 异常时保存部分输出并发射 error 事件
   - 无论成功失败，都发射 `task.completed` 或 `task.failed` 事件

2. **WebSocket 协程改为订阅者**：在 portal.py 的 ReAct 路径中：
   - 创建 task_id 并注册到 TaskStore（PENDING → RUNNING）
   - `asyncio.create_task(_execute_react_background(...))` 启动后台任务
   - `async for event in event_queue.subscribe(task_id)` 订阅事件
   - 将事件通过 `websocket.send_json()` 转发给前端
   - WebSocket 断开时，`async for` 循环退出，后台任务继续运行

3. **EventQueue 扩展**：当前 `subscribe()` 返回所有事件。需要添加按 `task_id` 过滤的订阅能力，或在前端转发时过滤。优先选择在 EventQueue 中添加 `subscribe(task_id=None)` 参数过滤。

4. **DIRECT_CHAT 路径**：DIRECT_CHAT 是同步 LLM 调用（非流式），保持现有逻辑不变，但同样在 TaskStore 注册。

**Technical design**（directional guidance）:

```python
# 后台执行函数（伪代码）
async def _execute_react_background(
    react_engine, messages, tools, model, agent_name,
    system_prompt, timeout_seconds,
    conv_id, task_id, event_queue, conversation_store
):
    collected_output = []
    try:
        async for event in react_engine.execute_stream(...):
            if event.event_type == "final_answer":
                collected_output.append(event.data.get("output", ""))
            await event_queue.emit(Event.create(
                event_type=event.event_type,
                task_id=task_id,
                session_id=conv_id,
                data={"step": event.step, "data": event.data, ...}
            ))
        # 正常完成
        response_text = _ensure_non_empty("".join(collected_output))
        await conversation_store.add_message(conv_id, "assistant", response_text)
        await event_queue.emit(Event.create(
            event_type="task.completed", task_id=task_id, ...
        ))
    except Exception as e:
        # 保存部分输出
        if collected_output:
            partial = _ensure_non_empty("".join(collected_output))
            await conversation_store.add_message(conv_id, "assistant", partial)
        await event_queue.emit(Event.create(
            event_type="task.failed", task_id=task_id,
            data={"error": str(e)}
        ))

# WebSocket 协程（伪代码）
task_id = str(uuid.uuid4())
task_store.create(task_id, agent_name, {"conversation_id": conv.id})
asyncio.create_task(_execute_react_background(...))

# WebSocket 订阅事件并转发
async for event in event_queue.subscribe(task_id=task_id):
    if event.event_type in ("task.completed", "task.failed"):
        await websocket.send_json({"type": "result", ...})
        break
    await websocket.send_json({"type": "step", "data": ...})
    # WebSocket 断开时 async for 退出，后台任务继续
```

**Patterns to follow**:
- EventQueue 的 `subscribe()` + 哨兵关闭模式（event_queue.py:193-243）
- `_emit_event_safe()` 的异常吞咽模式（portal.py:65-95）
- BackgroundRunner 的 `asyncio.create_task` + `_on_done` 回调模式（runner.py:55-73）

**Test scenarios**:
- **Happy path**: 后台任务正常完成，事件通过 EventQueue 分发，WebSocket 收到所有事件和最终结果
- **WebSocket 断开**: 后台任务继续运行，结果保存到 conversation store，TaskStore 状态更新为 COMPLETED
- **后台任务异常**: ReAct 执行抛异常，部分输出保存，TaskStore 状态为 FAILED，error 事件发射
- **EventQueue 订阅过滤**: 多个并发任务，每个 WebSocket 只收到自己 task_id 的事件
- **Integration**: 后台任务完成后，conversation store 有 assistant 消息，TaskStore 有 COMPLETED 记录

**Verification**: 启动后台任务后立即断开 WebSocket，等待任务完成后检查 conversation store 和 TaskStore。

---

### U3. Layer 2: TaskStore 注册与状态追踪

**Goal**: 在 portal WebSocket 路径中为每个用户消息创建 TaskStore 记录，状态随任务进展更新。

**Requirements**: R3

**Dependencies**: U2

**Files**:
- `src/agentkit/server/routes/portal.py` — 集成 TaskStore
- `tests/unit/test_portal_ws_task_tracking.py` — 新建测试

**Approach**:

1. 在 portal.py 的 WebSocket 路径中，获取 `task_store` from `websocket.app.state.task_store`。
2. 在用户消息处理开始时，调用 `task_store.create(task_id, agent_name, {"conversation_id": conv.id, "message": message_text})`。
3. 在后台任务执行前，更新状态为 `RUNNING`（`started_at=now`）。
4. 在后台任务完成时，更新状态为 `COMPLETED`（`output_data={"output": response_text}, completed_at=now, progress=1.0`）。
5. 在后台任务失败时，更新状态为 `FAILED`（`error_message=str(e), completed_at=now`）。
6. DIRECT_CHAT 路径同样注册 TaskStore（同步调用，状态直接从 PENDING → COMPLETED）。

**Patterns to follow**:
- BackgroundRunner._run_task 的状态更新模式（runner.py:89-92, 141-147, 153-157）
- TaskStore.create/update_status 的调用签名（task_store.py:127-163）

**Test scenarios**:
- **Happy path**: 用户发送消息 → TaskStore 创建 PENDING 记录 → 任务执行中状态为 RUNNING → 完成后状态为 COMPLETED，output_data 有内容
- **任务失败**: ReAct 执行异常 → TaskStore 状态为 FAILED，error_message 有内容
- **WebSocket 断开后查询**: WebSocket 断开，后台任务继续 → 通过 `GET /api/v1/tasks/{task_id}` 能查到 RUNNING 状态 → 任务完成后查到 COMPLETED
- **metadata 包含 conversation_id**: TaskStore 记录的 metadata 中有 conversation_id 字段
- **Integration**: 前端通过 `GET /api/v1/tasks?status=running` 能找到当前对话的运行中任务

**Verification**: 通过 REST API `GET /api/v1/tasks/{task_id}` 查询任务状态，验证状态流转正确。

---

### U4. Layer 3: 前端任务状态查询 API 客户端

**Goal**: 在前端 API 客户端中添加任务状态查询方法，支持按 status 过滤和按 task_id 查询。

**Requirements**: R5, R6

**Dependencies**: U3

**Files**:
- `src/agentkit/server/frontend/src/api/client.ts` — 添加任务 API 方法
- `src/agentkit/server/frontend/src/api/types.ts` — 添加任务类型定义

**Approach**:

1. 在 `types.ts` 中添加 `ITaskRecord` 接口，对应后端 `TaskRecord.to_dict()` 的输出格式。
2. 在 `client.ts` 中添加方法：
   - `getTask(taskId: string): Promise<ITaskRecord>` — GET `/api/v1/tasks/{taskId}`（注意：tasks 路由前缀是 `/api/v1/tasks`，不是 `/api/v1/portal`）
   - `listTasks(status?: string): Promise<ITaskRecord[]>` — GET `/api/v1/tasks?status=running`
3. 由于 tasks 路由前缀不同（`/api/v1/tasks` vs `/api/v1/portal`），需要创建一个新的 ApiClient 实例或调整 BaseApiClient 的 baseUrl。

**Patterns to follow**:
- 现有 `getConversations()` / `getConversation(id)` 的方法签名模式（client.ts:30-37）
- `IConversation` 接口定义模式（types.ts:51-57）

**Test scenarios**:
- **Happy path**: 调用 `getTask(taskId)` 返回正确的任务记录
- **按状态过滤**: 调用 `listTasks("running")` 只返回运行中任务
- **任务不存在**: 调用 `getTask("invalid-id")` 抛出 404 错误
- **Integration**: 后台任务运行中，前端能通过 API 查询到 RUNNING 状态

**Verification**: TypeScript 编译通过（`npm run build:frontend`），API 调用返回正确数据。

---

### U5. Layer 3: 前端 WebSocket 重连后的任务恢复

**Goal**: 前端 WebSocket 重连后，检查当前对话是否有未完成任务，恢复事件流或拉取已完成的结果。

**Requirements**: R5, R6

**Dependencies**: U4

**Files**:
- `src/agentkit/server/frontend/src/stores/chat.ts` — 添加重连恢复逻辑
- `src/agentkit/server/frontend/src/api/types.ts` — 扩展 WsClientMessage 类型

**Approach**:

1. **扩展 WebSocket 协议**：添加 `resume` 消息类型，前端重连后发送 `{type: "resume", task_id: "..."}` 订阅已有后台任务的事件流。
2. **后端处理 resume**：在 portal.py 的 WebSocket 路径中，处理 `resume` 消息类型——通过 task_id 订阅 EventQueue 事件流，转发给前端。
3. **前端重连恢复流程**（在 `connectWebSocket` 的 `onopen` 中）：
   - 检查 `currentConversationId` 是否有值
   - 调用 `listTasks("running")` 查找运行中任务
   - 匹配 `metadata.conversation_id === currentConversationId` 的任务
   - 如果找到运行中任务：发送 `resume` 消息，设置 `isLoading = true`
   - 如果无运行中任务：调用 `selectConversation(currentConversationId)` 重新加载消息（包含已完成的结果）
4. **后端 resume 处理**：接收到 `resume` 消息后，通过 `task_id` 订阅 EventQueue，转发事件直到 `task.completed` 或 `task.failed`。

**Technical design**（directional guidance）:

```typescript
// 前端重连恢复（伪代码）
socket.onopen = async () => {
  isWsConnected.value = true
  startHeartbeat()

  // 重连恢复逻辑
  if (currentConversationId.value) {
    await recoverTask(currentConversationId.value)
  }
}

async function recoverTask(convId: string) {
  const tasks = await apiClient.listTasks('running')
  const runningTask = tasks.find(
    t => t.metadata?.conversation_id === convId
  )

  if (runningTask) {
    // 恢复进行中任务的事件流
    isLoading.value = true
    ws.value.send(JSON.stringify({
      type: 'resume',
      task_id: runningTask.task_id
    }))
  } else {
    // 无运行中任务，重新加载对话消息
    await selectConversation(convId)
  }
}
```

**Patterns to follow**:
- 现有 `connectWebSocket` 的 `onopen` / `onclose` 模式（chat.ts:209-259）
- `selectConversation` 的消息加载模式（chat.ts:55-74）
- `handleWsMessage` 的事件处理模式（chat.ts:270-528）

**Test scenarios**:
- **Happy path - 有运行中任务**: 重连后发现有运行中任务 → 发送 resume → 收到后续事件 → 任务完成显示结果
- **Happy path - 无运行中任务**: 重连后无运行中任务 → 重新加载对话消息 → 显示已完成的结果
- **任务在重连前已完成**: 重连时任务已 COMPLETED → listTasks("running") 返回空 → 重新加载对话消息 → 结果显示
- **多个运行中任务**: 有多个对话的运行中任务 → 只恢复当前对话的任务
- **resume 后任务立即完成**: resume 后立即收到 task.completed 事件 → 正确显示结果
- **Integration**: 刷新页面 → 重连 → 恢复任务 → 最终结果正确显示

**Verification**: 启动复杂任务 → 刷新页面 → 验证任务继续运行 → 验证结果最终显示。

---

## Risks & Dependencies

### Risks

1. **EventQueue 订阅过滤改造风险**：当前 `subscribe()` 无过滤，添加 task_id 过滤可能影响现有订阅者。缓解：使用可选参数 `subscribe(task_id=None)`，默认行为不变。
2. **后台任务泄漏风险**：WebSocket 断开后后台任务继续运行，如果任务本身卡住（如 LLM 超时），任务会一直占用资源。缓解：ReAct 已有 `timeout_seconds` 配置，后台任务同样受此约束。
3. **并发任务事件混淆风险**：多个对话同时执行任务，EventQueue 事件可能混淆。缓解：每个任务有唯一 task_id，订阅时按 task_id 过滤。
4. **前端重连时序风险**：重连后查询任务状态时，任务可能刚好从 RUNNING 变为 COMPLETED。缓解：先查 running，如果空则重新加载对话消息（会包含已完成结果）。

### Dependencies

- U1 → U2 → U3 → U4 → U5（顺序依赖）
- EventQueue 已在 app.state 可用（无需新建）
- TaskStore 已在 app.state 可用（无需新建）

## System-Wide Impact

- **后端**：portal.py WebSocket 路径重大重构，影响所有 WebSocket 聊天用户
- **前端**：chat store 的 WebSocket 连接逻辑增强，影响所有聊天页面用户
- **API**：新增 `resume` WebSocket 消息类型，无 REST API 变更
- **兼容性**：DIRECT_CHAT 路径和 REST API 路径不受影响
