# 路线 01：前后端咨询与人工协作

## 学习场景

客户在消费者 H5 发送问题，页面持续接收处理状态与答复；客服在商家后台接管会话并发送人工回复。沿着消息、事件序号和控制版本，追踪浏览器与后端如何共同维护同一场会话。

图中后端路径相对 `src/zhifu/`；`frontend/`、`scripts/` 等路径相对仓库根目录。连续箭头表示函数直接调用；间断箭头注明 HTTP、框架回调或持久数据传递。框架节点、配置节点和数据节点单独标明。图中只展示本路线需要追踪的分支。返回[路线目录](README.md)。

## 图 1：客户消息进入会话事务

```mermaid
flowchart TD
    UI["frontend/src/pages/customer.js<br/>Vue 模板：van-form @submit"]
    SEND["frontend/src/pages/customer.js<br/>send() / perform(action)<br/>校验发送条件并执行 action"]
    ENSURE["frontend/src/pages/customer.js<br/>ensureConversation()<br/>读取或创建当前会话"]
    SERVICE["frontend/src/services/customer.js<br/>send() → request()<br/>附带正文与控制版本"]
    HTTP["frontend/src/api.js<br/>request()<br/>添加认证、幂等键并 fetch"]
    API["modules/conversations/api.py<br/>message()<br/>开启消息事务"]
    AUTH["modules/identity/auth.py<br/>authenticate()<br/>验证访问会话"]
    ACCESS["modules/conversations/service.py<br/>access()<br/>检查会话权限并锁定会话"]
    ACCEPT["modules/conversations/api.py<br/>accept_message() / prior()<br/>检查重复请求与控制版本"]
    QUEUE["modules/agent_runtime/service.py<br/>queue_run()<br/>创建待执行运行"]
    EMIT["modules/conversations/service.py<br/>emit()<br/>分配 sequence 并登记事件"]
    DB[("MySQL 持久数据<br/>AgentRun、Conversation<br/>ConversationEvent")]
    WORKER["modules/agent_runtime/worker.py<br/>claim()<br/>独立工作进程领取任务"]
    UI -. "Vue 调用提交处理函数" .-> SEND
    SEND -->|action 回调：查询当前状态| ENSURE
    SEND -->|action 回调：普通消息分支| SERVICE
    SERVICE --> HTTP
    HTTP -. "POST /openapi/v1/conversations/{id}/messages；FastAPI 路由" .-> API
    API --> AUTH
    API --> ACCESS
    API --> ACCEPT
    ACCEPT -->|客户处于 AGENT 模式| QUEUE
    ACCEPT --> EMIT
    QUEUE -. "同一事务登记运行" .-> DB
    EMIT -. "message 事务提交消息及会话变化" .-> DB
    DB -. "读取已提交的待执行运行" .-> WORKER
```

`send()` 的普通消息分支向会话接口提交消息。`ensureConversation()` 根据本地是否已有 `conversationId`，调用会话详情，或者查询会话列表并按需创建会话。`accept_message()` 先检查幂等记录，已有记录直接返回原事件；新请求继续检查 `expected_control_version`。

客户在 `AGENT` 模式下发送普通消息时，`queue_run()` 登记运行，当前会话保存 `active_run_id`。消息、运行和会话变化在 `message()` 的同一事务内提交。HTTP `201` 表示消息已被接受，后续答复通过图 2 的事件连接到达。`HUMAN` 模式客户消息与客服消息直接登记会话事件。

等待补充信息时，`send()` 调用 `services/customer.js` 的 `resume()`，请求 `/openapi/v1/agent/runs/{run_id}/inputs`。补充信息、模型调用和工具执行继续阅读 [路线 02：Agent 运行过程](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/docs/reading-routes/02-agent-runtime.md)。

### 图 1 源码导航

1. [send()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/pages/customer.js:254)、[perform()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/pages/customer.js:90)、[ensureConversation()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/pages/customer.js:136)：确认按钮条件、普通发送与补充信息分支。
2. [createCustomerService()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/services/customer.js:2)、[request()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/api.js:45)：观察接口路径、Bearer token、幂等键与错误处理。
3. [message()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/api.py:185)、[accept_message()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/api.py:205)、[prior()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/api.py:165)：定位事务、控制版本与幂等记录。
4. [authenticate()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/identity/auth.py:50)、[access()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/service.py:13)、[queue_run()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/service.py:20)、[emit()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/service.py:31)、[claim()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/agent_runtime/worker.py:28)：区分消息事务和后续工作进程。

## 图 2：历史读取、SSE 订阅与断线续传

```mermaid
flowchart TD
    WATCH["frontend/src/pages/customer.js<br/>Vue watch：props.page 变化"]
    LOAD["frontend/src/pages/customer.js<br/>loadPage()<br/>建立页面请求代次"]
    HISTORY["frontend/src/pages/customer.js<br/>refreshMessages()<br/>按 after 读取历史与当前运行"]
    CONNECT["frontend/src/pages/customer.js<br/>connectMessages()<br/>维护连接及续传循环"]
    CLIENT["frontend/src/services/customer.js<br/>events()<br/>选择当前会话事件地址"]
    FETCH["frontend/src/api.js<br/>events()<br/>fetch 与 EventSourceParserStream"]
    API["modules/conversations/stream.py<br/>events()<br/>检查权限、游标与 Redis 额度"]
    RESPONSE["modules/conversations/stream.py<br/>LeasedEventSourceResponse.__call__()<br/>管理响应及额度清理"]
    GENERATE["modules/conversations/stream.py<br/>generate()<br/>循环读取并生成 SSE 事件"]
    READ["modules/conversations/stream.py<br/>read_events()<br/>每轮重新验证身份与访问权限"]
    DB[("MySQL 持久数据<br/>ConversationEvent、Conversation<br/>会话身份及授权记录")]
    RECEIVE["frontend/src/pages/customer.js<br/>receiveEvent()<br/>处理序号、控制版本与页面状态"]
    RELEASE["modules/conversations/stream.py<br/>release()<br/>释放 Redis 连接额度"]
    WATCH -. "Vue 调用监听函数" .-> LOAD
    LOAD --> HISTORY
    LOAD -->|有效聊天页面且读取成功| CONNECT
    CONNECT --> CLIENT
    CLIENT --> FETCH
    FETCH -. "GET events；Last-Event-ID；FastAPI 路由" .-> API
    API -. "返回响应对象；ASGI 调用 __call__" .-> RESPONSE
    RESPONSE -. "EventSourceResponse 消费异步生成器" .-> GENERATE
    GENERATE --> READ
    DB -. "每轮短事务读取已提交事件" .-> READ
    GENERATE -. "HTTP SSE 正文：完整事件" .-> FETCH
    FETCH -->|调用传入的 onEvent 回调| RECEIVE
    CONNECT -->|410：本轮连接循环恢复一次| HISTORY
    GENERATE -->|finally| RELEASE
    RESPONSE -->|响应结束时 finally| RELEASE
```

`loadPage()` 关闭上一条连接，以 `generation` 和 `valid()` 限定当前页面请求。`refreshMessages()` 通过历史接口分页读取事件，并查询会话和活动运行；成功后 `connectMessages()` 建立 SSE。`api.events()` 使用依赖 `EventSourceParserStream` 读取完整事件，随后调用页面传入的 `onEvent`。

服务端每轮 `read_events()` 使用独立事务验证 token、账户、租户和会话权限，查询完成后释放数据库连接。`generate()` 推送事件的 `id`、`event` 和 JSON `data`；当前传输单位为完整持久事件。每次连接约 50 秒结束，客户端正常结束后携带 `state.after` 继续订阅。

`receiveEvent()` 跳过已处理序号，再更新 `state.after`；旧控制版本的助手、Agent 进度和运行事件随后被过滤。游标过期时，连接循环自动重新读取历史和当前状态一次。网络异常显示重新连接入口；页面切换或组件卸载会中止旧连接。`release()` 同时由生成器与响应生命周期保护，重复调用保持幂等。

### 图 2 源码导航

1. [loadPage() 与 watch 注册](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/pages/customer.js:373)、[refreshMessages()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/pages/customer.js:152)：跟踪初始化、历史分页和页面代次。
2. [connectMessages()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/pages/customer.js:216)、[客户端 events()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/api.js:11)、[receiveEvent()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/pages/customer.js:177)：跟踪续传游标及 UI 状态。
3. [服务端 events()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/stream.py:84)、[generate()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/stream.py:131)、[read_events()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/stream.py:50)：检查订阅与每轮查询的边界。
4. [LeasedEventSourceResponse.__call__()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/stream.py:41)、[release()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/stream.py:125)、[history()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/api.py:135)：核对提前断开、额度清理和历史记录读取。

## 图 3：客服接管、人工回复与消费者页面更新

```mermaid
flowchart TD
    CONTROL["frontend/src/pages/merchant.js<br/>control(action)<br/>请求确认接管或交还"]
    CONFIRM["frontend/src/main.js<br/>confirm()<br/>保存确认窗口与操作回调"]
    ACCEPT["frontend/src/App.js<br/>accept()<br/>确认后执行已保存回调"]
    SERVICE["frontend/src/services/merchant.js<br/>control() / command() / request()<br/>附带 expected_control_version"]
    HTTP["frontend/src/api.js<br/>request()<br/>发送商家身份请求"]
    API["modules/conversations/api.py<br/>control()<br/>权限、幂等和控制版本检查"]
    EMIT["modules/conversations/service.py<br/>emit()<br/>登记 control.changed"]
    REPLY["frontend/src/pages/merchant.js<br/>reply()<br/>发送人工回复并刷新会话"]
    REPLY_SERVICE["frontend/src/services/merchant.js<br/>reply() / command() / request()<br/>附带会话控制版本"]
    MESSAGE["modules/conversations/api.py<br/>message() / accept_message()<br/>确认当前接管客服并登记回复"]
    REFRESH["frontend/src/services/merchant.js<br/>conversation()<br/>重新查询详情与消息历史"]
    QUERY["modules/conversations/api.py<br/>get_conversation() / history()<br/>读取可访问的会话与事件"]
    DB[("MySQL 持久数据<br/>Conversation、AgentRun<br/>ConversationEvent")]
    SSE["modules/conversations/stream.py<br/>read_events() / generate()<br/>已有 SSE 连接读取新事件"]
    CUSTOMER["frontend/src/pages/customer.js<br/>receiveEvent()<br/>显示控制变化与人工消息"]
    CONTROL --> CONFIRM
    CONFIRM -. "Vue 展示窗口；用户确认触发 accept" .-> ACCEPT
    ACCEPT -->|执行回调中的 run 与 action| SERVICE
    SERVICE --> HTTP
    HTTP -. "POST conversations/{id}/{action}；FastAPI 路由" .-> API
    API --> EMIT
    EMIT -. "事务提交控制变化与适用的运行取消" .-> DB
    REPLY -->|run 的 action 回调| REPLY_SERVICE
    REPLY_SERVICE --> HTTP
    HTTP -. "POST conversations/{id}/messages；FastAPI 路由" .-> MESSAGE
    MESSAGE -. "同一事务提交 message.staff" .-> DB
    ACCEPT -->|控制请求成功后，回调继续调用| REFRESH
    REPLY -->|回复成功后，action 回调继续调用| REFRESH
    REFRESH -->|经服务 request 包装| HTTP
    HTTP -. "GET 会话详情与 messages；FastAPI 路由" .-> QUERY
    DB -. "事务查询已提交记录" .-> QUERY
    DB -. "读取已提交的控制变化与人工消息" .-> SSE
    SSE -. "HTTP SSE，经图 2 客户端处理" .-> CUSTOMER
```

`control()` 要求 `conversations:manage` 权限；接管成功后将控制方设为 `HUMAN`，增加 `control_version`，记录 `assigned_staff_id`，并取消适用的活动运行。交还要求当前员工就是已接管客服。交还之后，客户下一次普通发言创建新的 Agent 运行。

客服回复要求 `conversations:reply` 权限、`HUMAN` 控制方，以及匹配的 `assigned_staff_id`。商家页面打开会话、执行刷新、回复成功或控制操作成功时读取详情与历史。消费者通过持续 SSE 接收 `message.staff`；商家通知的 30 秒轮询负责通知和未读数量。

### 图 3 源码导航

1. [页面 control() 与 reply()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/pages/merchant.js:126)、[confirm()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/main.js:140)、[accept()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/App.js:64)、[服务 control() 与 reply()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/services/merchant.js:62)：核对确认步骤、请求字段和成功后刷新。
2. [后端 control()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/api.py:264)、[accept_message()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/api.py:205)：定位接管、交还和人工回复的权限检查。
3. [conversation()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/src/services/merchant.js:44)、[get_conversation()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/api.py:122)、[history()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/api.py:135)：核对商家页面读取的数据。
4. [finish_run()](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/src/zhifu/modules/conversations/service.py:96)：理解运行提交时对控制方、控制版本和活动运行的检查。

## 关键数据与状态

- `conversation_id` 标识会话；`active_run_id` 指向当前活动运行。客户普通消息被接受后，后端可能创建新的 `AgentRun`。
- `sequence` 为会话持久事件序号；`state.after` 为前端已经处理到的序号。历史分页、SSE 续传和消息去重使用这些字段。
- `control_owner` 为 `AGENT` 或 `HUMAN`；`control_version` 在接管、交还时增加。请求中的 `expected_control_version` 保护当前操作，事件中的版本帮助页面识别旧回复。
- `Idempotency-Key` 与消息摘要共同限定同一请求。会话消息及控制操作的作用域为租户、会话和操作者。
- `stream_cursor` 表示可续传窗口边界；游标过期使用 `410 EVENT_CURSOR_EXPIRED`。持久历史继续由 `messages` 接口提供。
- `WAITING_USER`、`WAITING_APPROVAL`、`run.waiting`、`run.resumed` 支持补充信息与业务等待；`run.failed`、`run.cancelled` 使页面清理对应活动运行。

## 阅读检查问题

1. 同一条消息由于网络原因重复提交时，前后端分别在哪里保留或检查幂等标识？为何返回 `201` 后页面仍可能收到 `run.failed`？
2. `state.after`、`sequence`、`stream_cursor` 各自控制什么？游标过期与普通连接结束分别经过哪些函数？
3. 客服接管时，后端怎样取消已有运行，消费者页面怎样处理迟到的助手事件，客服回复又受到哪些身份条件限制？

## 测试文件阅读入口

本路线只阅读用例及其断言；编写本文未执行这些用例，也未调用生成模型。

- [test_conversations.py](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/tests/test_conversations.py:47)：接管与交还、并发接管、消息幂等、SSE 续传、权限失效与保留窗口。
- [test_sse_limits.py](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/tests/test_sse_limits.py:18)：Redis 原子连接额度与共享限制。
- [customer-sse.mjs](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/tests/customer-sse.mjs:53)：浏览器接收人工消息、正常续传、断网恢复、页面离开及游标过期。
- [merchant-conversation.mjs](G:/leon/miaoa/workspace/shuh-ai/zhifu-platform/frontend/tests/merchant-conversation.mjs:56)：商家队列进入、人工接管、回复与客户历史可见性。