agent-Specialization/docs/analysis-task-recovery.md

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

概述

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

---

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

1.1 恢复触发时机

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

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

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()

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()

// 检查是否需要从头重建
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 思考块处理

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 工具意图打字机效果

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 消息开始处理

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()

@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()

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()

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()

// 恢复思考块状态
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 工具块状态恢复

// 注册历史中的工具块到 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 停止按钮状态

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

状态设置:

• streamingMessage = true: 显示停止按钮
• taskInProgress = true: 禁用输入框
• stopRequested = false: 重置停止请求标志

4.4 输入框状态

输入框状态由 taskInProgress 控制：

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

状态:

• taskInProgress = true: 输入框禁用
• taskInProgress = false: 输入框启用

---

5. 轮询的恢复机制

5.1 轮询启动

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

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 从头重建模式

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

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

5.2.2 增量恢复模式

// 获取任务详情，计算已处理的事件数量
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

// 处理新事件
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. 状态同步: 恢复思考块、工具块、文本块的完整状态

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