# AI Agent 系统任务恢复与推送机制分析

## 概述

本文档详细分析 AI Agent 系统中**正在运行中的对话恢复机制和推送机制**，重点关注页面刷新后如何恢复任务状态、重放历史事件以及维护 UI 一致性。

---

## 1. 页面刷新后的任务恢复流程

### 1.1 恢复触发时机

任务恢复在页面挂载时立即触发：

**文件**: `/static/src/app/lifecycle.ts`

```typescript
export async function mounted() {
    // ... 其他初始化代码

    // 注册全局事件处理器（用于任务轮询）
    (window as any).__taskEventHandler = (event: any) => {
        if (typeof this.handleTaskEvent === 'function') {
            this.handleTaskEvent(event);
        }
    };

    // 立即尝试恢复运行中的任务（不延迟）
    if (typeof this.restoreTaskState === 'function') {
        this.restoreTaskState();
    }
}
```

**关键点**:
- 在 `mounted()` 生命周期钩子中立即调用 `restoreTaskState()`
- 不延迟执行，确保用户刷新后能快速看到任务恢复
- 先注册全局事件处理器 `__taskEventHandler`，用于后续轮询事件分发

### 1.2 查找运行中的任务

**文件**: `/static/src/stores/task.ts` - `loadRunningTask()`

```typescript
async loadRunningTask(conversationId: string | null = null) {
    try {
        debugLog('[Task] 查找运行中的任务');

        // 1. 获取所有任务列表
        const response = await fetch('/api/tasks');
        const result = await response.json();

        // 2. 查找运行中的任务（可选按对话ID过滤）
        const runningTask = result.data.find((task: any) =>
            task.status === 'running' &&
            (!conversationId || task.conversation_id === conversationId)
        );

        if (runningTask) {
            // 3. 更新任务状态
            this.currentTaskId = runningTask.task_id;
            this.taskStatus = runningTask.status;
            this.taskCreatedAt = runningTask.created_at;
            this.taskUpdatedAt = runningTask.updated_at;

            // 4. 获取任务详情，计算已处理的事件数量
            const detailResponse = await fetch(`/api/tasks/${runningTask.task_id}`);
            if (detailResponse.ok) {
                const detailResult = await detailResponse.json();
                if (detailResult.success && detailResult.data.events) {
                    // 设置为当前事件数量，只获取新事件
                    this.lastEventIndex = detailResult.data.next_offset || detailResult.data.events.length;
                    debugLog('[Task] 设置起始偏移量:', this.lastEventIndex);
                }
            }

            return runningTask;
        }

        return null;
    } catch (error) {
        console.error('[Task] 加载运行中任务失败:', error);
        return null;
    }
}
```

**关键逻辑**:
1. 调用 `/api/tasks` 获取用户的所有任务
2. 筛选 `status === 'running'` 且匹配当前对话ID的任务
3. 获取任务详情，计算 `lastEventIndex`（已处理的事件数量）
4. 返回运行中的任务信息

---

## 2. `_rebuildingFromScratch` 标志机制

### 2.1 标志的作用

`_rebuildingFromScratch` 是一个关键的布尔标志，用于区分两种恢复模式：

- **从头重建模式** (`true`): 历史消息不完整或缺失，需要从头重放所有事件来重建 assistant 响应
- **增量恢复模式** (`false`): 历史消息完整，只需恢复状态并继续轮询新事件

### 2.2 何时设置为 true

**文件**: `/static/src/app/methods/taskPolling.ts` - `restoreTaskState()`

```typescript
// 检查是否需要从头重建
const historyActionsCount = lastMessage?.actions?.length || 0;
const eventCount = allEvents.length;
const historyIncomplete = eventCount > historyActionsCount + 5; // 允许5个事件的误差

const needsRebuild = !isAssistantMessage ||
                     (isAssistantMessage && (!lastMessage.actions || lastMessage.actions.length === 0)) ||
                     historyIncomplete;

if (needsRebuild) {
    debugLog('[TaskPolling] 需要从头重建 assistant 响应');

    // 清空不完整的 assistant 消息
    if (isAssistantMessage) {
        this.messages.pop();
    }

    // 重置偏移量为 0，从头获取所有事件
    taskStore.lastEventIndex = 0;

    // 标记正在从头重建
    this._rebuildingFromScratch = true;
    this._rebuildingEventCount = allEvents.length;

    // 启动轮询
    taskStore.startPolling((event: any) => {
        this.handleTaskEvent(event);
    });

    // 延迟清除重建标记（2秒后）
    setTimeout(() => {
        this._rebuildingFromScratch = false;
        this._rebuildingEventCount = 0;
    }, 2000);
}
```

**触发条件**:
1. **最后一条消息不是 assistant 消息**: 说明历史加载有问题
2. **最后一条是空的 assistant 消息**: `actions` 数组为空或不存在
3. **历史不完整**: 事件数量远大于历史中的 actions 数量（允许5个事件的误差）

### 2.3 对事件处理的影响

`_rebuildingFromScratch` 标志会影响多个事件处理函数的行为：

#### 2.3.1 思考块处理

```typescript
handleThinkingStart(data: any, eventIdx: number) {
    // ... 创建思考块

    // 只在非历史恢复时展开思考块
    if (!this._rebuildingFromScratch) {
        this.chatExpandBlock(blockId);
    }
}

handleThinkingEnd(data: any) {
    // ... 完成思考块

    // 只在非历史恢复时延迟折叠思考块
    if (!this._rebuildingFromScratch) {
        setTimeout(() => {
            this.chatCollapseBlock(blockId);
        }, 1000);
    } else {
        // 历史恢复时立即折叠，不需要动画
        this.chatCollapseBlock(blockId);
    }
}
```

**差异**:
- **从头重建**: 不展开思考块，立即折叠，跳过动画
- **正常流式**: 展开思考块，延迟1秒后折叠，有动画效果

#### 2.3.2 工具意图打字机效果

```typescript
handleToolIntent(data: any) {
    const isHistoryRestore = this._rebuildingFromScratch;

    if (isHistoryRestore) {
        // 历史恢复，直接显示完整 intent
        action.tool.intent_rendered = newIntent;
    } else {
        // 新工具块，逐字符显示（打字机效果）
        action.tool.intent_rendered = '';
        action.tool._intentTyping = true;

        // 逐字符显示动画
        const typeNextChar = () => {
            if (charIndex < newIntent.length && action.tool._intentTyping) {
                action.tool.intent_rendered += newIntent[charIndex];
                charIndex++;
                this.$forceUpdate();
                action.tool._intentTimer = setTimeout(typeNextChar, charInterval);
            }
        };
        action.tool._intentTimer = setTimeout(typeNextChar, 50);
    }
}
```

**差异**:
- **从头重建**: 直接显示完整 intent，无打字机效果
- **正常流式**: 逐字符显示，有打字机动画

#### 2.3.3 AI 消息开始处理

```typescript
handleAiMessageStart(data: any, eventIdx: number) {
    // 如果是从头重建，标记消息为静默恢复
    if (this._rebuildingFromScratch) {
        const newMessage = this.messages[this.messages.length - 1];
        if (newMessage && newMessage.role === 'assistant') {
            newMessage.awaitingFirstContent = false;
            newMessage.generatingLabel = '';
        }
    }
}
```

**差异**:
- **从头重建**: 清除"生成中"标签，静默恢复
- **正常流式**: 显示"生成中"动画

---

## 3. 历史事件的重放机制

### 3.1 事件获取

**后端**: `/server/tasks.py` - `get_task_api()`

```python
@tasks_bp.route("/api/tasks/<task_id>", methods=["GET"])
@api_login_required
def get_task_api(task_id: str):
    username = get_current_username()
    rec = task_manager.get_task(username, task_id)
    if not rec:
        return jsonify({"success": False, "error": "任务不存在"}), 404

    # 获取偏移量参数
    try:
        offset = int(request.args.get("from", 0))
    except Exception:
        offset = 0

    # 过滤事件（只返回 idx >= offset 的事件）
    events = [e for e in rec.events if e["idx"] >= offset]
    next_offset = events[-1]["idx"] + 1 if events else offset

    return jsonify({
        "success": True,
        "data": {
            "task_id": rec.task_id,
            "status": rec.status,
            "events": events,
            "next_offset": next_offset,
        }
    })
```

**关键点**:
- 支持 `from` 参数，只返回 `idx >= from` 的事件
- 返回 `next_offset`，用于下次轮询
- 事件存储在 `deque(maxlen=1000)` 中，最多保留1000个事件

### 3.2 事件重放顺序

**文件**: `/static/src/stores/task.ts` - `pollTaskEvents()`

```typescript
async pollTaskEvents(eventHandler: (event: any) => void) {
    const response = await fetch(
        `/api/tasks/${this.currentTaskId}?from=${this.lastEventIndex}`
    );
    const result = await response.json();
    const data = result.data;

    // 处理新事件（按顺序）
    if (data.events && data.events.length > 0) {
        for (const event of data.events) {
            try {
                eventHandler(event);
            } catch (err) {
                console.error('[Task] 处理事件失败:', err, event);
            }
        }

        // 更新偏移量
        this.lastEventIndex = data.next_offset;
    }
}
```

**重放顺序**:
1. 按事件的 `idx` 顺序（从小到大）
2. 同步处理，确保顺序正确
3. 每个事件处理完后才处理下一个

### 3.3 重放时的特殊处理

**文件**: `/static/src/app/methods/taskPolling.ts` - `handleTaskEvent()`

```typescript
handleTaskEvent(event: any) {
    // 1. 检查事件的 conversation_id 是否匹配
    if (eventData.conversation_id && this.currentConversationId) {
        if (eventData.conversation_id !== this.currentConversationId) {
            debugLog(`忽略不匹配的事件`);
            return;
        }
    }

    // 2. 根据事件类型调用对应的处理方法
    switch (eventType) {
        case 'ai_message_start':
            this.handleAiMessageStart(eventData, eventIdx);
            break;
        case 'thinking_start':
            this.handleThinkingStart(eventData, eventIdx);
            break;
        // ... 其他事件类型
    }

    // 注意：不要在这里清除 _rebuildingFromScratch 标记
    // 该标记应该在恢复完所有历史事件后才清除
}
```

**特殊处理**:
- **对话ID过滤**: 忽略不匹配当前对话的事件
- **跳过动画**: 通过 `_rebuildingFromScratch` 标志跳过动画效果
- **静默恢复**: 不显示"生成中"标签，不展开思考块

---

## 4. 恢复时的 UI 状态管理

### 4.1 思考块状态恢复

**文件**: `/static/src/app/methods/taskPolling.ts` - `restoreTaskState()`

```typescript
// 恢复思考块状态
if (lastMessage.actions) {
    const thinkingActions = lastMessage.actions.filter(a => a.type === 'thinking');

    if (inThinking && thinkingActions.length > 0) {
        // 正在思考中，检查最后一个思考块是否正在流式输出
        const lastThinking = thinkingActions[thinkingActions.length - 1];

        // 只有当思考块正在流式输出时才设置锁定状态（但不展开）
        if (lastThinking.streaming && lastThinking.blockId) {
            this.$nextTick(() => {
                this.chatSetThinkingLock(lastThinking.blockId, true);
            });
        }
    }

    // 确保所有思考块都是折叠状态
    for (const thinking of thinkingActions) {
        if (thinking.blockId) {
            thinking.collapsed = true;
        }
    }
}
```

**状态恢复**:
- **折叠状态**: 所有思考块默认折叠
- **锁定状态**: 如果正在流式输出，设置锁定（防止用户手动展开）
- **不展开**: 即使正在流式输出，也不自动展开（避免干扰用户）

### 4.2 工具块状态恢复

```typescript
// 注册历史中的工具块到 toolActionIndex
const toolActions = lastMessage.actions.filter(a => a.type === 'tool');

for (const toolAction of toolActions) {
    if (toolAction.tool && toolAction.tool.id) {
        // 注册到 toolActionIndex
        this.toolRegisterAction(toolAction, toolAction.tool.id);

        // 如果有 executionId，也注册
        if (toolAction.tool.executionId) {
            this.toolRegisterAction(toolAction, toolAction.tool.executionId);
        }

        // 追踪工具调用
        if (toolAction.tool.name) {
            this.toolTrackAction(toolAction.tool.name, toolAction);
        }
    }
}
```

**关键点**:
- **注册工具块**: 将历史中的工具块注册到 `toolActionIndex`
- **支持更新**: 后续的 `update_action` 事件可以找到对应的块进行状态更新
- **双重注册**: 同时注册 `id` 和 `executionId`，确保能找到

### 4.3 停止按钮状态

```typescript
// 标记状态为进行中
this.streamingMessage = true;
this.taskInProgress = true;
```

**状态设置**:
- `streamingMessage = true`: 显示停止按钮
- `taskInProgress = true`: 禁用输入框
- `stopRequested = false`: 重置停止请求标志

### 4.4 输入框状态

输入框状态由 `taskInProgress` 控制：

```typescript
// 在模板中
<input :disabled="taskInProgress" />
```

**状态**:
- `taskInProgress = true`: 输入框禁用
- `taskInProgress = false`: 输入框启用

---

## 5. 轮询的恢复机制

### 5.1 轮询启动

**文件**: `/static/src/stores/task.ts` - `startPolling()`

```typescript
startPolling(eventHandler?: (event: any) => void) {
    if (this.isPolling) {
        return;
    }

    this.isPolling = true;

    // 获取事件处理器
    const handler = eventHandler || ((window as any).__taskEventHandler);

    // 立即执行一次
    this.pollTaskEvents(handler);

    // 设置定时轮询（150ms 间隔）
    this.pollingInterval = window.setInterval(() => {
        this.pollTaskEvents(handler);
    }, 150);
}
```

**关键点**:
- **立即执行**: 启动轮询后立即执行一次，不等待第一个间隔
- **150ms 间隔**: 接近流式输出效果
- **全局处理器**: 使用 `window.__taskEventHandler` 作为默认处理器

### 5.2 从哪个 offset 开始轮询

**两种模式**:

#### 5.2.1 从头重建模式

```typescript
// 重置偏移量为 0，从头获取所有事件
taskStore.lastEventIndex = 0;
```

- **offset = 0**: 从第一个事件开始
- **目的**: 重建完整的 assistant 响应

#### 5.2.2 增量恢复模式

```typescript
// 获取任务详情，计算已处理的事件数量
const detailResponse = await fetch(`/api/tasks/${runningTask.task_id}`);
const detailResult = await detailResponse.json();
if (detailResult.success && detailResult.data.events) {
    // 设置为当前事件数量，只获取新事件
    this.lastEventIndex = detailResult.data.next_offset || detailResult.data.events.length;
}
```

- **offset = next_offset**: 从历史事件之后开始
- **目的**: 只获取新事件，避免重复处理

### 5.3 避免重复处理事件

**机制**:

1. **事件索引**: 每个事件有唯一的 `idx`
2. **偏移量过滤**: 后端只返回 `idx >= offset` 的事件
3. **偏移量更新**: 处理完事件后更新 `lastEventIndex = next_offset`

```typescript
// 处理新事件
if (data.events && data.events.length > 0) {
    for (const event of data.events) {
        eventHandler(event);
    }

    // 更新偏移量，下次轮询从这里开始
    this.lastEventIndex = data.next_offset;
}
```

**保证**:
- 每个事件只处理一次
- 不会遗漏事件
- 不会重复处理

---

## 6. WebSocket vs 轮询的恢复差异

### 6.1 WebSocket 模式（已废弃）

**特点**:
- **实时推送**: 服务器主动推送事件
- **连接断开**: 刷新页面后 WebSocket 连接断开
- **无法恢复**: 断开期间的事件丢失

**问题**:
- 页面刷新后无法恢复任务状态
- 需要复杂的重连和事件补偿机制

### 6.2 轮询模式（当前实现）

**特点**:
- **主动拉取**: 客户端主动轮询事件
- **无连接状态**: 不依赖持久连接
- **完整恢复**: 可以从任意 offset 开始拉取事件

**优势**:
- 页面刷新后可以完整恢复任务状态
- 实现简单，无需处理重连逻辑
- 支持从头重建和增量恢复两种模式

### 6.3 恢复流程对比

| 特性 | WebSocket 模式 | 轮询模式 |
|------|---------------|---------|
| 连接状态 | 有状态（需要重连） | 无状态 |
| 事件恢复 | 困难（需要补偿机制） | 简单（从 offset 拉取） |
| 实时性 | 高（服务器推送） | 中（150ms 轮询） |
| 可靠性 | 低（连接断开丢失事件） | 高（事件持久化） |
| 刷新恢复 | 不支持 | 完全支持 |

---

## 7. 完整恢复流程图

```
页面刷新
    ↓
mounted() 生命周期
    ↓
注册全局事件处理器 __taskEventHandler
    ↓
调用 restoreTaskState()
    ↓
检查是否已有任务在进行 ──→ 是 ──→ 跳过恢复
    ↓ 否
调用 taskStore.loadRunningTask()
    ↓
查询 /api/tasks 获取任务列表
    ↓
筛选 status === 'running' 的任务 ──→ 无 ──→ 恢复子智能体等待状态
    ↓ 有
获取任务详情 /api/tasks/{task_id}
    ↓
计算 lastEventIndex（已处理的事件数量）
    ↓
等待历史加载完成
    ↓
检查历史完整性
    ↓
    ├─→ 历史不完整 ──→ 从头重建模式
    │       ↓
    │   删除不完整的 assistant 消息
    │       ↓
    │   设置 _rebuildingFromScratch = true
    │       ↓
    │   设置 lastEventIndex = 0
    │       ↓
    │   启动轮询（从头重放所有事件）
    │       ↓
    │   2秒后清除 _rebuildingFromScratch 标志
    │
    └─→ 历史完整 ──→ 增量恢复模式
            ↓
        恢复思考块状态（折叠、锁定）
            ↓
        恢复工具块状态（注册到 toolActionIndex）
            ↓
        恢复文本块状态（streaming 标志）
            ↓
        设置 streamingMessage = true
            ↓
        设置 taskInProgress = true
            ↓
        启动轮询（只获取新事件）
            ↓
        显示"任务恢复"提示
```

---

## 8. 关键代码文件索引

### 8.1 前端文件

| 文件路径 | 功能 |
|---------|------|
| `/static/src/app/lifecycle.ts` | 页面生命周期，触发恢复 |
| `/static/src/stores/task.ts` | 任务状态管理，轮询控制 |
| `/static/src/app/methods/taskPolling.ts` | 事件处理，状态恢复 |
| `/static/src/app/methods/history.ts` | 历史消息加载和渲染 |
| `/static/src/app/methods/conversation.ts` | 对话管理 |

### 8.2 后端文件

| 文件路径 | 功能 |
|---------|------|
| `/server/tasks.py` | 任务 API，事件存储和查询 |
| `/server/chat_flow.py` | 聊天流程，事件生成 |

---

## 9. 最佳实践和注意事项

### 9.1 事件设计

1. **每个事件必须有唯一的 idx**: 用于偏移量过滤
2. **事件必须包含 conversation_id**: 用于过滤不匹配的事件
3. **事件必须按顺序生成**: 确保 idx 递增
4. **事件必须持久化**: 存储在 `deque(maxlen=1000)` 中

### 9.2 状态恢复

1. **先加载历史，再恢复任务**: 确保有完整的消息列表
2. **检查历史完整性**: 对比事件数量和 actions 数量
3. **注册工具块**: 确保后续的 `update_action` 事件能找到对应的块
4. **设置正确的偏移量**: 避免重复处理或遗漏事件

### 9.3 UI 体验

1. **跳过动画**: 从头重建时跳过所有动画效果
2. **静默恢复**: 不显示"生成中"标签
3. **折叠思考块**: 默认折叠，避免干扰用户
4. **显示提示**: 恢复成功后显示"任务恢复"提示

### 9.4 错误处理

1. **任务不存在**: 恢复子智能体等待状态
2. **历史加载失败**: 延迟重试（最多5次）
3. **事件处理失败**: 捕获异常，继续处理下一个事件
4. **对话切换**: 忽略不匹配的事件

---

## 10. 总结

AI Agent 系统的任务恢复机制通过以下关键技术实现：

1. **事件持久化**: 所有事件存储在后端，支持从任意 offset 查询
2. **轮询模式**: 客户端主动拉取事件，无需维护连接状态
3. **双模式恢复**: 支持从头重建和增量恢复两种模式
4. **智能判断**: 根据历史完整性自动选择恢复模式
5. **UI 优化**: 跳过动画、静默恢复，提升用户体验
6. **状态同步**: 恢复思考块、工具块、文本块的完整状态

这套机制确保了用户在页面刷新后能够无缝恢复任务，不会丢失任何进度，同时保持了良好的用户体验。