zhifu-platform/docs/reading-routes/07-channels-messaging.md

路线 7：外部渠道与可靠消息

代码依据：2026-09-30 当前工作区。学习场景：外部渠道发送客户咨询或确认事件；平台保存业务结果，再向 OMS 和原渠道传递事件。

图中的后端路径相对 src/zhifu/。连续箭头表示当前进程内的调用或标明条件的执行步骤；间断箭头说明 HTTP、消息队列、数据库记录与后续请求的连接。节点中的 内部 表示该函数内的逻辑。返回路线目录。

图 7A：签名验证、身份关联与入站去重

flowchart TD
    A0["外部渠道<br/>客户消息、补充资料或方案确认"]
    A1["7A-1 modules/channels/api.py<br/>receive<br/>HMAC 入站入口"]
    A2["7A-2 modules/channels/auth.py<br/>authenticate_channel<br/>原始正文、绑定、时间窗、签名与 nonce"]
    A3["7A-3 modules/channels/standard.py<br/>receive_standard → authenticate_standard<br/>由 Standard Webhooks SDK 验签"]
    A4["7A-4 modules/channels/standard_schemas.py<br/>to_envelope<br/>转换业务事件"]
    A5["7A-5 modules/channels/service.py<br/>accept_envelope<br/>检查 scope、ChannelIdentity 与当前客户权限"]
    A6{"同一 external_message_id<br/>已经保存 ChannelInbox？"}
    A7["accept_envelope 内部<br/>同客户、同内容返回原收据<br/>内容冲突终止请求"]
    A8["accept_envelope 内部<br/>建立内部 AccessSession，取得或创建渠道会话"]
    A9["7A-9 modules/conversations/api.py<br/>accept_message<br/>保存消息；AGENT 模式同时创建 AgentRun"]
    A10["7A-10 modules/agent_runtime/inputs.py<br/>accept_input<br/>将补充资料交给原等待运行"]
    A11["7A-11 modules/after_sales/service.py<br/>AfterSalesService.execute → confirm<br/>登记客户确认"]
    A12["accept_envelope 返回<br/>ChannelInbox 与业务变更共用事务<br/>路由提交后返回 HTTP 202"]
    A13["Agent worker<br/>后续读取已提交的运行"]
    A14["ConversationEvent<br/>HUMAN 模式消息供客服读取<br/>接管与回复见路线 1"]
    A15["售后确认记录与 BusinessEvent<br/>业务等待的后续核对见路线 2"]
    A0 -.->|HMAC HTTP| A1
    A1 --> A2
    A2 -->|验签返回后，receive 调用| A5
    A0 -.->|Standard Webhooks HTTP| A3
    A3 -->|验签成功| A4
    A4 -->|receive_standard 调用| A5
    A5 --> A6
    A6 -->|已经保存| A7
    A6 -->|首次受理| A8
    A8 -->|message| A9
    A8 -->|clarification| A10
    A8 -->|proposal_confirmation| A11
    A9 --> A12
    A10 --> A12
    A11 --> A12
    A12 -.->|AGENT 模式新任务或补充输入恢复任务| A13
    A12 -.->|HUMAN 模式消息，run_id 为 null| A14
    A12 -.->|方案确认事件| A15

源码导航：

• 7A-1、7A-2：receive、authenticate_channel。继续阅读 raw_request 与 load_binding。
• 7A-3、7A-4：receive_standard、authenticate_standard、to_envelope。
• 7A-5 至 7A-8、7A-12：accept_envelope。寻找权限检查与 ChannelInbox 查询的先后顺序，以及内部 AccessSession 的用途。
• 7A-9：accept_message，会话处理见路线 1。
• 7A-10：accept_input，运行恢复见路线 2。
• 7A-11：AfterSalesService.execute，客户确认与主管审批见路线 3。

两种协议都先验证当前绑定，再由已登记的 ChannelIdentity 找到客户。HMAC 使用时间窗和 ChannelNonce 防止同一签名重复使用；业务重复发送使用原 external_message_id、原内容和新 nonce。Standard Webhooks 使用 webhook-id 作为业务消息标识，交给同一 Inbox 流程去重。

客户身份、店铺权限、业务状态和幂等收据在当前请求事务中检查。普通消息、补充资料和方案确认分别进入对应业务函数。客户消息处于 AGENT 模式时创建运行，HUMAN 模式只保存消息并返回 run_id=null。HTTP 202 表示已经受理并提交记录；运行与履约由各自后续流程推进。

图 7B：业务事件、RabbitMQ 与消费收据

flowchart TD
    B1["7B-1 modules/after_sales/service.py<br/>AfterSalesService.decide<br/>批准售后申请"]
    B2["7B-2 modules/after_sales/service.py<br/>changed → emit<br/>业务修改与 BusinessEvent 共用事务"]
    B3["MySQL BusinessEvent<br/>event_type=after_sale.approved<br/>delivery_status=PENDING"]
    B4["7B-4 modules/integrations/worker.py<br/>main 内部 publisher → publish_one<br/>领取 Outbox 事件租约"]
    B5["7B-5 modules/integrations/worker.py<br/>Broker.publish_event<br/>EventEnvelope 与持久消息"]
    B6["RabbitMQ<br/>发布确认与持久队列"]
    B7["publish_one 返回处理<br/>验证原租约后登记 SENT"]
    B8["7B-8 modules/integrations/worker.py<br/>main 内部 consumer → consume_one<br/>读取消息并校验 schema_version"]
    B9["7B-9 modules/integrations/fulfillment.py<br/>prepare_event<br/>检查 ConsumerReceipt，按事件准备业务"]
    B10["7B-10 modules/integrations/fulfillment.py<br/>prepare<br/>复验审批、订单、规则与数量"]
    B11["同一事务<br/>ConsumerReceipt + ExternalOperation + Reservation<br/>有效批准事件进入履约准备"]
    B12["consume_one 的 message.process 退出<br/>事务成功后确认消息<br/>异常请求重新投递"]
    B13["7B-13 modules/integrations/fulfillment.py<br/>execute_due<br/>executor 读取已准备的操作"]
    B1 -->|批准分支| B2
    B2 -.->|调用方事务提交| B3
    B3 -.->|publisher 查询数据库| B4
    B4 --> B5
    B5 -.->|AMQP 发布| B6
    B6 -.->|确认返回| B7
    B6 -.->|consumer 读取| B8
    B8 --> B9
    B9 -->|批准事件且没有消费收据| B10
    B10 -->|验证成功，返回后保存收据| B11
    B9 -->|已有收据，结束本次业务处理| B12
    B11 -->|提交后返回 consume_one| B12
    B11 -.->|独立 executor 查询数据库| B13

源码导航：

• 7B-1、7B-2：AfterSalesService.decide、changed、emit。
• 7B-4、7B-5、7B-8：main、publish_one、Broker.publish_event、consume_one。main 通过 TaskGroup 分别运行 publisher、consumer 和 executor。
• 7B-9、7B-10、7B-13：prepare_event、prepare、execute_due。OMS 网络调用与结果查询继续阅读路线 3。

BusinessEvent 同时承担 Outbox 记录。publish_one 先领取租约，在网络发布成功后用原租约检查提交 SENT。发布成功后进程中断，事件可能再次发送；ConsumerReceipt 使重复消费能够读取已经处理的结果。

consume_one 按 event_id 回到数据库读取业务事实。prepare_event 在适用的批准事件中查询 OMS 订单，再进入事务检查收据和最新业务状态。过期事件、已经处理的事件与当前有效事件具有不同分支；图 7B 的 ExternalOperation 与数量预占属于有效批准事件。

发布者保存 SENT 与消费者保存收据由各自任务调度，没有固定先后关系。运维核对入口为 message_recovery.reconcile：检查审批事件的发送状态、有效租约与消费收据；提交操作人和原因后，仅对 SENT_WITHOUT_RECEIPT 重新登记 PENDING，保留原 event_id。

图 7C：业务结果返回外部渠道

flowchart TD
    C1["7C-1 modules/conversations/service.py<br/>emit<br/>webhook 会话保存指定回复与控制事件"]
    C2["7C-2 modules/after_sales/service.py<br/>emit<br/>保存售后业务事件"]
    C3["MySQL BusinessEvent<br/>conversation.* 或 after_sale.*"]
    C4["7C-4 modules/platform/delivery.py<br/>main → enqueue<br/>按渠道 scope 查询事件"]
    C5["MySQL ChannelDelivery<br/>唯一事件投递记录，QUEUED"]
    C6["7C-6 modules/platform/delivery.py<br/>execute_one<br/>检查前序投递、当前授权及租约"]
    C7["execute_one 内部<br/>会话事件复验 ChannelIdentity 与 control_version"]
    C8["execute_one 内部<br/>提交 SENDING、attempts、lease_token<br/>准备 HMAC 请求"]
    C9["渠道 callback_url<br/>httpx.AsyncClient.post"]
    C10["execute_one 返回处理<br/>原租约有效，2xx 登记 SENT"]
    C11["execute_one 返回处理<br/>错误登记 QUEUED 或 FAILED"]
    C12["7C-12 modules/platform/api.py<br/>retry<br/>平台员工检查版本并重新排队"]
    C1 -.->|事务提交| C3
    C2 -.->|事务提交| C3
    C3 -.->|数据库扫描| C4
    C4 -.->|保存投递记录| C5
    C5 -.->|main 随后调用 execute_one 领取| C6
    C6 -->|会话事件| C7
    C6 -->|工单事件| C8
    C7 -->|身份及控制版本有效| C8
    C8 -.->|签名 HTTP POST| C9
    C9 -.->|2xx| C10
    C9 -.->|网络错误或非 2xx| C11
    C11 -.->|QUEUED 到达 next_attempt_at 后读取| C6
    C11 -.->|FAILED 后员工发起独立请求| C12
    C12 -.->|保存 QUEUED，等待后续领取| C6

源码导航：

• 7C-1：conversations.service.emit。只有 webhook 会话中的指定事件会同时创建渠道用 BusinessEvent。
• 7C-2：AfterSalesService.emit，履约更新还使用 integrations.fulfillment.emit。
• 7C-4、7C-6 至 7C-11：delivery.main、enqueue、execute_one。
• 7C-12：platform.api.retry。沿 service.command 继续阅读平台操作的幂等控制。

渠道投递工作进程直接扫描数据库的业务事件；它与 RabbitMQ 履约消费各自执行。相同会话或工单存在较早版本的 QUEUED、SENDING 投递时，后续版本暂时不能取得发送租约。

每组三次自动发送，失败后登记 FAILED；平台员工可以提交独立重试请求。每次 HTTP 使用 5 秒客户端超时，外层期限为 6 秒。会话身份映射或控制版本变化、渠道停用、缺少签名凭据也会进入失败处理。图 7C 展示成功授权后的发送主线。

发送受理在取得会话锁时登记；其后发生的接管不能撤销已经受理的 HTTP 请求。返回后仍需核对 lease_token 与 SENDING，迟到结果只能提交到原租约仍然有效的投递记录。

关键数据与状态

• ChannelKey、ChannelNonce、ChannelInbox：签名密钥、签名重复使用检查、业务去重收据。
• BusinessEvent：业务版本、Outbox 发送状态与租约。
• ConsumerReceipt、ExternalOperation、Reservation：消息已经处理的证据、外部操作与数量预占。
• ChannelDelivery：QUEUED、SENDING、SENT、FAILED，以及发送次数、版本与租约。

阅读检查

1. 找到入站签名标识、业务消息标识与内部命令幂等键，分别说明重复发送时哪些值保持相同、哪些值重新生成。
2. 从审批事务追踪 BusinessEvent、发布确认与 ConsumerReceipt，解释在各个位置发生进程中断时，后续任务怎样判断是否继续执行。
3. 跟踪一条 conversation.message.assistant 事件到 ChannelDelivery，说明人工接管前后哪些检查能够改变发送结果，以及发送受理之后的边界。

测试阅读入口

以下文件用于追踪预期行为；本文编写时只阅读用例，没有执行其中的业务请求。

• test_channel_inbox.py、test_standard_webhooks.py：入站签名、身份与去重。
• test_fulfillment.py、message_recovery_acceptance.py：履约收据、消息恢复与重复处理。
• test_channel_acceptance.py：渠道业务与结果投递。