App-server、exec-server 和 relay 传输通道

fn app_server_control_socket_path(codex_home: &Path) -> std::io::Result<AbsolutePathBuf>

作用：拼出应用服务器控制用 Unix socket 文件的位置。Unix socket 可以理解成本机程序之间通信的一根“本地电话线”。

数据流：输入是 Codex 的主目录路径 → 它在这个目录下面追加固定的子目录名和 socket 文件名 → 输出一个确认过的绝对路径；如果路径不是合法绝对路径，就返回错误。

调用关系：当用户写 unix:// 但没有指定具体 socket 路径时，AppServerTransport::from_listen_url 会调用它来决定默认 socket 放在哪里。它只负责算路径，不负责创建文件或监听连接。

调用图：调用 1 个内部函数（from_absolute_path）；被 1 处调用（from_listen_url）；外部调用 1 个（join）。

fn app_server_startup_lock_path(codex_home: &Path) -> std::io::Result<AbsolutePathBuf>

作用：拼出应用服务器启动锁文件的位置。启动锁像门口的一把锁，用来防止多个服务器实例同时抢着启动。

数据流：输入是 Codex 的主目录路径 → 它拼上固定的控制目录和锁文件名 → 输出一个绝对路径，或者在路径不合规时返回错误。

调用关系：主启动流程 run_main_with_transport_options 会用它找到锁文件位置，然后再由别的代码去真正加锁。这个函数只提供统一的文件地址。

调用图：调用 1 个内部函数（from_absolute_path）；被 1 处调用（run_main_with_transport_options）；外部调用 1 个（join）。

fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result

作用：把“监听地址解析失败”的错误变成人能看懂的文字。这样命令行报错时，用户知道自己该写什么格式。

数据流：输入是某一种解析错误 → 它根据错误类型写出不同说明，比如不支持的 URL、Unix socket 路径无效、WebSocket 地址格式不对 → 输出给错误显示系统一段提示文字。

调用关系：它是 AppServerTransportParseError 这个错误类型的显示方式。AppServerTransport::from_listen_url 产生错误后，日志、命令行或测试在展示错误时会间接用到它。

调用图：外部调用 1 个（write!）。

fn from_listen_url(listen_url: &str) -> Result<Self, AppServerTransportParseError>

作用：把用户写的监听地址字符串变成程序内部能使用的传输方式。比如 stdio://、unix://...、ws://IP:PORT 或 off。

数据流：输入是一段 listen URL 字符串 → 它按前缀判断要用标准输入输出、本机 Unix socket、WebSocket，还是关闭服务；需要默认 socket 时会先找到 Codex 主目录再拼路径；需要自定义路径时会转成绝对路径；格式不对就返回清楚的解析错误 → 输出一个 AppServerTransport 枚举值。

调用关系：这是启动配置进入传输层的关键入口。测试 explicit_remote_control_startup_fails_when_disabled_by_requirements 会用它构造传输配置；它内部会把默认 Unix socket 路径的工作交给 app_server_control_socket_path。

调用图：调用 3 个内部函数（app_server_control_socket_path, find_codex_home, relative_to_current_dir）；被 1 处调用（explicit_remote_control_startup_fails_when_disabled_by_requirements）；外部调用 1 个（UnsupportedListenUrl）。

fn from_str(s: &str) -> Result<Self, Self::Err>

作用：让 AppServerTransport 可以直接从字符串解析出来，方便和 Rust 标准的字符串解析习惯配合使用。

数据流：输入是一段字符串 → 它直接交给 AppServerTransport::from_listen_url 解析 → 输出传输方式或解析错误。

调用关系：这是一个薄薄的适配层。真正判断格式的活儿都交给 AppServerTransport::from_listen_url，它让外部代码可以用通用的 FromStr 方式解析配置。

调用图：外部调用 1 个（from_listen_url）。

fn next_connection_id() -> ConnectionId

作用：给新连接发一个不会重复的编号。这样后面收到消息、关闭连接、回发消息时，都知道是在说哪一个连接。

数据流：它不需要外部输入 → 从全局的原子计数器里取当前数字并加一；原子计数器可以理解成多任务同时访问也不会数错的计数器 → 输出一个 ConnectionId。

调用关系：各个具体连接接入代码会在新客户端连上时使用它。它不关心连接来自标准输入、WebSocket 还是远程控制，只负责发号。

调用图：外部调用 1 个（new）。

async fn forward_incoming_message(
    transport_event_tx: &mpsc::Sender<TransportEvent>,
    writer: &mpsc::Sender<QueuedOutgoingMessage>,
    connection_id: ConnectionId,
    payload: &str,
) -> boo

作用：把外面传进来的原始文字消息解析成 JSON-RPC 消息，然后转交给内部队列。它是“文字入口”到“结构化事件”的转换点。

数据流：输入是传输事件发送器、当前连接的写回通道、连接编号和一段字符串 payload → 它先尝试把字符串解析成 JSONRPCMessage；解析成功就调用 enqueue_incoming_message 放进服务器事件队列；解析失败就记一条错误日志并继续保持连接 → 输出 true 或 false，表示后续是否还能继续处理。

调用关系：具体传输模块读到客户端消息后会走到这里。它不自己决定队列满了怎么办，而是把这部分交给 enqueue_incoming_message。

调用图：调用 1 个内部函数（enqueue_incoming_message）；外部调用 1 个（error!）。

async fn enqueue_incoming_message(
    transport_event_tx: &mpsc::Sender<TransportEvent>,
    writer: &mpsc::Sender<QueuedOutgoingMessage>,
    connection_id: ConnectionId,
    message: JSONRPCMessage

作用：把已经解析好的客户端消息塞进服务器内部事件队列，并在队列太满时做保护。它避免服务器因为一波请求涌入而无限占内存或卡死。

数据流：输入是内部事件队列、给该连接写回消息的队列、连接编号和 JSON-RPC 消息 → 它先尝试立刻放入内部队列；如果队列关闭，就返回 false；如果队列满且消息是请求，就尝试给客户端回一个“服务器过载，稍后重试”的错误；如果队列满但消息不是请求，比如响应，就等待队列有空位再放进去 → 输出布尔值，并可能向写回队列新增一条错误回复。

调用关系：forward_incoming_message 会调用它处理所有入站消息。多个测试专门验证它在队列满时的行为：请求会收到过载错误，响应不会被丢掉，写回队列满时也不会把系统拖住。

调用图：调用 1 个内部函数（new）；被 3 处调用（forward_incoming_message, enqueue_incoming_request_does_not_block_when_writer_queue_is_full, enqueue_incoming_response_waits_instead_of_dropping_when_queue_is_full）；外部调用 4 个（send, try_send, Error, warn!）。

fn serialize_outgoing_message(outgoing_message: OutgoingMessage) -> Option<String>

作用：把服务器准备发出去的消息变成 JSON 字符串，供标准输入输出、socket 或 WebSocket 发送。

数据流：输入是一个 OutgoingMessage → 它先转成 JSON 值，再转成真正的字符串；任何一步失败都会写错误日志 → 输出 Some(JSON 字符串)，失败时输出 None。

调用关系：它位于出站消息的最后加工阶段。具体传输写出数据前会需要这种字符串形式；它不负责发送，只负责把消息打包成可传输文本。

调用图：外部调用 3 个（error!, to_string, to_value）。

fn listen_off_parses_as_off_transport()

作用：确认用户写 off 时，传输层会被解析成关闭状态。这个测试防止以后有人改解析逻辑时破坏这个开关。

数据流：输入是字符串 off → 调用解析函数得到结果 → 检查结果是不是 AppServerTransport::Off。

调用关系：它验证 AppServerTransport::from_listen_url 的一个重要分支：完全关闭监听。这个分支对不想启动服务器通道的场景很关键。

调用图：外部调用 1 个（assert_eq!）。

async fn enqueue_incoming_request_returns_overload_error_when_queue_is_full()

作用：测试内部事件队列满时，新来的请求会收到“服务器太忙”的错误，而不是被静默丢掉。

数据流：它先创建容量很小的内部队列和写回队列 → 先塞入一条消息把内部队列占满 → 再提交一个请求 → 检查原来的消息还在内部队列里，同时写回队列里出现了带指定错误码的过载错误。

调用关系：它直接验证 enqueue_incoming_message 的过载保护。这个测试说明系统宁愿明确告诉客户端稍后重试，也不让请求在内部排队无限膨胀。

调用图：外部调用 10 个（Notification, Request, Integer, new, assert!, assert_eq!, json!, channel, panic!, to_value）。

async fn enqueue_incoming_response_waits_instead_of_dropping_when_queue_is_full()

作用：测试内部队列满时，响应消息会等待，而不是像新请求那样被过载处理。响应通常对应已经发生过的请求，丢掉会让双方状态对不上。

数据流：它先把内部队列填满 → 在另一个异步任务里尝试塞入一个响应 → 再从队列取走旧消息腾出位置 → 检查响应最终成功进入队列，内容和连接编号都正确。

调用关系：它调用 enqueue_incoming_message，专门覆盖“满队列 + 响应消息”的路径。这个测试保证队列压力下也不会轻易丢掉已有对话的回复。

调用图：调用 1 个内部函数（enqueue_incoming_message）；外部调用 10 个（Notification, Response, Integer, new, assert!, assert_eq!, json!, channel, panic!, spawn）。

async fn enqueue_incoming_request_does_not_block_when_writer_queue_is_full()

作用：测试当服务器想回“过载错误”但写回队列也满了时，函数不会一直卡住。这样一个慢客户端不会拖垮处理流程。

数据流：它把内部事件队列填满，再把写回队列也填满 → 提交一个新请求并设置很短的超时时间 → 确认 enqueue_incoming_message 很快返回，并且写回队列原本的消息没有被挤掉。

调用关系：它验证 enqueue_incoming_message 的另一个保护点：连错误回复都发不出去时，就记录警告并继续，而不是等待到天荒地老。

调用图：调用 2 个内部函数（new, enqueue_incoming_message）；外部调用 13 个（from_millis, ConfigWarning, Notification, Request, Integer, new, AppServerNotification, assert!, assert_eq!, json! (+3 more)）。

fn observe(&mut self, envelope: ClientEnvelope) -> ClientSegmentObservation

作用：接收一个客户端信封，判断它是普通消息还是“大消息的一小段”。普通消息直接放行；分段消息则检查、保存、拼接，拼完整后再变回普通消息放行。

数据流：输入是一封 ClientEnvelope，也就是带客户端编号、流编号、事件内容的消息包。它先看事件是不是 ClientMessageChunk；如果不是，就原样输出 Forward。若是分段，它读取段号、总段数、总大小、base64 内容、seq_id 和 stream_id，检查有没有缺字段、超大小、乱序、重复、编码错误。通过检查后，它把这一段解码成原始字节，追加到对应客户端的临时拼装区。还没收齐时输出 Pending；发现问题时删除拼装区并输出 Dropped；全部收齐且能解析成 JSONRPCMessage 时，输出一封新的普通 ClientMessage。

调用关系：它是客户端分段消息进入系统后的主要关口，由 observe_client_message 调用。它会向 ClientSegmentMetadata::from_envelope 取分段元信息，用 should_ignore_chunk 判断旧段是否该忽略，用 evict_assemblies_if_full 防止临时拼装区太多，用 remove_assembly 在失败或完成后清理。它的结果决定后续流程是继续等待、丢弃，还是把完整消息往下传。

调用图：调用 4 个内部函数（from_envelope, evict_assemblies_if_full, remove_assembly, should_ignore_chunk）；被 1 处调用（observe_client_message）；外部调用 8 个（new, now, new, Complete, Forward, decoded_len_estimate, min, warn!）。

fn invalidate_stream(&mut self, client_id: &ClientId, stream_id: &StreamId)

作用：当某个客户端的某条流失效时，清掉这条流正在拼的半截大消息。这样旧流留下的碎片不会混进新流里。

数据流：输入是 client_id 和 stream_id。它检查当前是否有这个客户端的拼装记录，并且记录里的流编号是否匹配；匹配就删除，不匹配就不动。输出没有返回值，变化是内部的临时拼装缓存可能被清掉。

调用关系：它由 observe_client_message 在发现流需要作废时调用。实际删除动作交给 remove_assembly，因为 remove_assembly 会额外确认流编号，避免误删同一客户端其他流的拼装数据。

调用图：调用 1 个内部函数（remove_assembly）；被 1 处调用（observe_client_message）。

fn invalidate_client(&mut self, client_id: &ClientId)

作用：当某个客户端整体失效或断开时，清掉这个客户端所有未拼完的大消息。这样不会让离线客户端的残留数据一直占内存。

数据流：输入是 client_id。它直接从内部表里删除这个客户端对应的拼装记录。输出没有返回值，结果是这个客户端的半成品消息不再保留。

调用关系：这是客户端生命周期变化时用的清理入口。和 invalidate_stream 不同，它不关心具体 stream_id，因为整个客户端都已经不该继续保留状态了。

fn should_ignore_chunk(
        &self,
        client_id: &ClientId,
        stream_id: &StreamId,
        seq_id: u64,
        segment_id: usize,
    ) -> bool

作用：判断某个分段是不是已经过时或已经收过的旧片段。这样重复到达的网络包不会让拼装流程倒退或报错。

数据流：输入是 client_id、stream_id、seq_id 和 segment_id。它查看内部是否已经在为这个客户端、这条流拼装消息；如果正在拼，并且传来的序号比当前消息旧，或者同一条消息里段号小于下一个应收段号，就返回 true。否则返回 false。它只读取状态，不修改任何东西。

调用关系：observe 在真正处理分段前会先问它一次，observe_client_message 也会直接用它来判断旧块。它相当于收货员看快递单号：已经签收过的箱子就别再搬进仓库。

调用图：被 2 处调用（observe, observe_client_message）。

fn remove_assembly(&mut self, client_id: &ClientId, stream_id: &StreamId)

作用：安全删除某个客户端、某条流的临时拼装记录。它会先确认流编号对得上，避免删错。

数据流：输入是 client_id 和 stream_id。它查内部拼装表，如果这个客户端确实有记录，且记录的 stream_id 和输入一致，就删除这条记录；否则什么也不做。输出没有返回值，可能改变内部缓存。

调用关系：它是多个清理路径共用的小工具。observe 在分段失败或拼完后调用它，invalidate_stream 也调用它。这样所有“结束这次拼装”的动作都走同一个安全出口。

调用图：被 2 处调用（invalidate_stream, observe）。

fn evict_assemblies_if_full(&mut self)

作用：当同时等待拼装的大消息太多时，踢掉最久没有新片段的那一个。这样可以防止大量半截消息把内存占满。

数据流：它不接收外部参数，只查看内部 assemblies 表。如果表里的拼装数量达到上限，就反复找到 last_chunk_seen_at 最早的记录，也就是最久没动静的记录，然后删除，直到数量低于上限。没有返回值，改变的是内部缓存大小。

调用关系：observe 在为一个新客户端创建拼装区前会调用它。它不参与拼消息内容，只负责“仓库满了先清旧货”，保证后面的新分段有地方放。

调用图：被 1 处调用（observe）。

fn from_envelope(envelope: &ClientEnvelope) -> Option<Self>

作用：从客户端信封里提取分段拼装所需的关键信息。没有这些信息，就没法判断哪些小段属于同一条大消息。

数据流：输入是一封 ClientEnvelope。它先确认事件类型确实是 ClientMessageChunk，然后读取 segment_count、message_size_bytes，并从信封上取 seq_id。成功时输出 ClientSegmentMetadata，里面有消息序号、总段数和总大小；如果不是分段消息或缺少 seq_id，就输出 None。

调用关系：observe 在处理每个客户端分段时调用它。它给 observe 提供“这批箱子的订单信息”，后续才能检查元信息是否一致、是否旧消息、是否该重置拼装。

调用图：被 1 处调用（observe）。

fn split_server_envelope_for_transport(
    envelope: ServerEnvelope,
) -> io::Result<Vec<ServerEnvelope>>

作用：在服务器往外发消息前，检查这封消息会不会太大；太大就切成多封小消息。这样每次发出去的包都尽量落在传输能接受的大小内。

数据流：输入是一封 ServerEnvelope。它先看事件是不是 ServerMessage；不是就原样放进列表返回。是普通服务器消息时，它用 serialized_len 估算整封信封编码成 JSON 后有多大；不超过上限就原样返回。超过上限时，它把真正的 message 转成字节数组，检查总大小，再计算合适的切块数量和每块大小。它反复试算每个分块信封的序列化大小，直到所有块都能放进限制内，然后输出多个 ServerEnvelope，每个事件都是 ServerMessageChunk。如果消息太大或怎么切都装不下，就输出空列表，等于丢弃这条消息并记录警告。

调用关系：它由 run_server_writer_inner 在真正写出服务器消息前调用，测试 splits_large_server_messages_into_wire_chunks 也会验证它。它内部用 serialized_len 和 serialized_chunk_len 做大小估算，用 build_chunk_envelope 生成每一个分段信封。它是服务器出站方向的分箱工。

调用图：调用 2 个内部函数（serialized_chunk_len, serialized_len）；被 2 处调用（splits_large_server_messages_into_wire_chunks, run_server_writer_inner）；外部调用 8 个（new, matches!, to_vec, unreachable!, max, min, vec!, warn!）。

fn serialized_chunk_len(
    envelope: &ServerEnvelope,
    segment_id: usize,
    segment_count: usize,
    message_size_bytes: usize,
    chunk: &[u8],
) -> io::Result<usize>

作用：估算某个服务器消息分块打包成信封后，实际写成 JSON 会有多大。这个大小决定分块能不能安全发送。

数据流：输入是原始 ServerEnvelope、当前段号、总段数、原消息总字节数，以及这一段的原始字节。它先用 build_chunk_envelope 把这一段包装成 ServerMessageChunk 信封，再用 serialized_len 计算这个信封序列化后的字节数。输出是大小数字，或者在构造失败时返回错误。

调用关系：它只被 split_server_envelope_for_transport 调用，用于试算不同切法是否合格。它自己不决定怎么切，只回答“这一块装箱后有多大”。

调用图：调用 2 个内部函数（build_chunk_envelope, serialized_len）；被 1 处调用（split_server_envelope_for_transport）。

fn write(&mut self, buf: &[u8]) -> io::Result<usize>

作用：这是一个只数字节、不真正保存内容的写入器。它让程序能知道 JSON 写出来会有多长，而不用真的生成一份字符串。

数据流：输入是一段要写入的字节 buf。它把 buf 的长度加到自己的 len 计数上，然后报告“这些字节都写成功了”。输出是写入的字节数，内部变化是 len 变大。

调用关系：它实现了 Write 接口，也就是 Rust 里“可以被写入”的通用约定。serialized_len 会把它交给 serde_json::to_writer，让 JSON 序列化过程像正常写文件一样写入，但实际只是在计数。

fn flush(&mut self) -> io::Result<()>

作用：完成写入器接口要求的刷新动作。因为 CountingWriter 不真的缓存内容，所以刷新时什么也不用做。

数据流：它没有输入数据，也不修改状态，直接返回成功。结果表示“刷新完成”。

调用关系：它和 CountingWriter::write 一起满足 Write 接口。serde_json::to_writer 可能按接口要求调用刷新；这里返回成功即可。

fn serialized_len(value: &impl serde::Serialize) -> io::Result<usize>

作用：计算一个可序列化对象写成 JSON 后的字节长度。它用来提前判断消息会不会超过传输大小限制。

数据流：输入是任意实现 serde::Serialize 的值，也就是能被转成 JSON 的数据。它创建一个 CountingWriter，然后让 serde_json::to_writer 把这个值写进去。CountingWriter 不保存内容，只累计写入字节数。最后输出累计长度；如果转 JSON 失败，就输出 io 错误。

调用关系：split_server_envelope_for_transport 用它检查整封服务器信封大小，serialized_chunk_len 用它检查单个分块信封大小。它是本文件所有“先量尺寸再决定怎么发”的基础工具。

调用图：被 2 处调用（serialized_chunk_len, split_server_envelope_for_transport）；外部调用 2 个（default, to_writer）。

fn build_chunk_envelope(
    envelope: &ServerEnvelope,
    segment_id: usize,
    segment_count: usize,
    message_size_bytes: usize,
    chunk: &[u8],
) -> io::Result<ServerEnvelope>

作用：把服务器消息的一小段原始字节包装成可以发送的分段信封。它会带上段号、总段数、原消息大小等信息，方便接收端以后拼回去。

数据流：输入是原始 ServerEnvelope、segment_id、segment_count、message_size_bytes 和当前 chunk 字节。它先检查总段数不能超过上限；超过就返回 InvalidData 错误。通过后，它把 chunk 用 base64 编成文本，放进 ServerEvent::ServerMessageChunk，同时复制原信封里的 client_id、stream_id 和 seq_id。输出是一封新的 ServerEnvelope。

调用关系：serialized_chunk_len 用它临时构造分段信封来量大小；split_server_envelope_for_transport 最终也依赖这套构造方式生成实际要发的分段。它是服务器出站分块时的“装箱”步骤。

调用图：被 1 处调用（serialized_chunk_len）；外部调用 1 个（new）。

async fn list_remote_control_clients(
    remote_control_url: &str,
    auth_manager: &Arc<AuthManager>,
    params: RemoteControlClientsListParams,
) -> io::Result<RemoteControlClientsListResponse>

作用：这个函数用来获取某个远程控制环境里的客户端列表，比如哪些手机、桌面端或其他设备曾经连接过。它会先替调用方把环境编号、分页数量这些输入检查好，避免发出明显错误的请求。

数据流：进去的是远程控制服务地址、认证管理器，以及列表查询参数，比如 environment_id、cursor、limit、order。它先确认 environment_id 不为空，limit 在 1 到 100 之间；然后用 environment_clients_url 拼出客户端列表网址；接着通过 send_client_management_request 发网络请求；回来后用 ensure_success_response 检查 HTTP 状态是否成功；最后把服务器 JSON 里的 items 转成 RemoteControlClient 列表，把 cursor 变成 next_cursor 返回。它不直接改业务数据，只返回查询结果或错误。

调用关系：它通常被更上层的 list_clients 调用，也被相关测试用来确认解析错误和认证恢复是否正确。它自己不负责底层网络细节，而是把网址拼接交给 environment_clients_url，把发请求和认证重试交给 send_client_management_request，把失败判断交给 ensure_success_response，把响应正文预览交给 preview_remote_control_response_body。

调用图：调用 4 个内部函数（ensure_success_response, environment_clients_url, send_client_management_request, preview_remote_control_response_body）；被 4 处调用（list_clients, list_remote_control_clients_preserves_decode_error_context, list_remote_control_clients_recovers_auth_after_unauthorized, list_remote_control_clients_retries_unauthorized_only_once）；外部调用 1 个（new）。

async fn revoke_remote_control_client(
    remote_control_url: &str,
    auth_manager: &Arc<AuthManager>,
    params: RemoteControlClientsRevokeParams,
) -> io::Result<RemoteControlClientsRevokeRespon

作用：这个函数用来撤销某个远程控制客户端，相当于把指定设备从允许连接名单里移除。它能防止上层忘记传环境编号或客户端编号后还把错误请求发出去。

数据流：进去的是远程控制服务地址、认证管理器，以及撤销参数，里面必须有 environment_id 和 client_id。它先检查这两个值都不为空；再拼出环境的 clients 地址，并把 client_id 追加到网址末尾；然后通过 send_client_management_request 发 DELETE 请求；收到响应后预览正文并用 ensure_success_response 判断是否成功；成功时返回一个空的 RemoteControlClientsRevokeResponse，表示撤销动作已经被服务器接受。

调用关系：它通常被上层的 revoke_client 调用，也有测试确认遇到 403 这类无权限错误时不会乱重试。它负责撤销流程的入口判断和结果检查，真正的网络发送、认证头添加和 401 后重试由 send_client_management_request 继续完成。

调用图：调用 4 个内部函数（ensure_success_response, environment_clients_url, send_client_management_request, preview_remote_control_response_body）；被 2 处调用（revoke_client, revoke_remote_control_client_does_not_retry_forbidden）；外部调用 1 个（new）。

async fn send_client_management_request(
    auth_manager: &Arc<AuthManager>,
    request: ClientManagementRequest<'_>,
    action: &str,
) -> io::Result<ClientManagementResponse>

作用：这个函数是客户端管理请求的“带认证发送器”。它不关心是列表还是撤销，重点是确保请求带着当前登录凭证，并在凭证过期时尝试恢复一次。

数据流：进去的是认证管理器、一个请求描述，以及一段动作说明文字。它先从认证管理器拿到认证恢复工具和认证变化监听器，再加载当前远程控制认证信息；然后调用 send_client_management_request_once 真正发一次请求。如果返回状态不是 401，或者认证恢复失败，它就直接把这次响应交回去；如果是 401 且恢复成功，它会重新加载认证，再发第二次。出来的是一次最终的 HTTP 响应包装，里面有状态码、响应头和响应正文。

调用关系：list_remote_control_clients 和 revoke_remote_control_client 都通过它发送请求，因为这两个操作都需要同样的登录处理。它把“认证可能过期”的复杂情况挡在中间层，底层单次发送交给 send_client_management_request_once，认证加载和恢复分别交给 load_remote_control_auth 与 recover_remote_control_auth。

调用图：调用 3 个内部函数（load_remote_control_auth, recover_remote_control_auth, send_client_management_request_once）；被 2 处调用（list_remote_control_clients, revoke_remote_control_client）。

async fn send_client_management_request_once(
    auth: &RemoteControlConnectionAuth,
    request: &ClientManagementRequest<'_>,
    action: &str,
) -> io::Result<ClientManagementResponse>

作用：这个函数只负责把某一次 HTTP 请求真正发出去，不做多轮重试。可以把它理解成“按当前凭证打一通电话给服务器”。

数据流：进去的是已经准备好的远程控制认证信息、请求类型，以及动作说明文字。它创建网络客户端，生成认证头；如果是 List 请求，就构造 GET 请求并附上 cursor、limit、order 这些查询参数；如果是 Revoke 请求，就构造 DELETE 请求。然后设置 30 秒超时，带上认证头和账号编号发出请求。出来的是 ClientManagementResponse，包含 HTTP 状态、响应头和完整响应正文；如果网络失败或正文读取失败，就返回带说明的错误。

调用关系：它只被 send_client_management_request 调用，所以它的位置很靠近底层网络。上层是否需要因为 401 再试一次，不由它决定；它只负责一次发送、一次接收，并把原始结果带回去。

调用图：调用 1 个内部函数（build_reqwest_client）；被 1 处调用（send_client_management_request）；外部调用 3 个（new, new, timeout）。

fn ensure_success_response(
    status: axum::http::StatusCode,
    headers: &HeaderMap,
    url: &Url,
    body_preview: &str,
    response_kind: &str,
) -> io::Result<()>

作用：这个函数用来统一判断远程控制服务器的响应是不是成功。失败时，它会把不同 HTTP 状态码翻译成更容易被程序处理的错误类型。

数据流：进去的是 HTTP 状态码、响应头、请求网址、响应正文预览，以及响应类型说明。它先看状态码是不是成功范围；如果是，就什么也不改，直接返回成功。否则它根据状态码选择错误类别：400 算输入错，401 或 403 算权限不够，404 算找不到，其他算普通错误。出来的是一个包含网址、状态码、响应头和正文预览的错误，方便人和程序判断问题。

调用关系：list_remote_control_clients 和 revoke_remote_control_client 在拿到服务器响应后都会调用它。它是这两个公开操作的统一“验收员”：只有它放行，后面才会解析列表结果或确认撤销成功。

调用图：被 2 处调用（list_remote_control_clients, revoke_remote_control_client）；外部调用 4 个（as_u16, is_success, new, format!）。

fn environment_clients_url(remote_control_url: &str, environment_id: &str) -> io::Result<Url>

作用：这个函数负责把基础远程控制地址和环境编号拼成“这个环境的客户端管理接口”地址。它避免每个调用方自己手写路径，减少拼错网址的风险。

数据流：进去的是远程控制服务的基础地址字符串和 environment_id。它先用 normalize_remote_control_base_url 把基础地址整理成标准形式，再追加固定路径 wham/remote/control/environments，最后继续追加环境编号和 clients。出来的是一个 Url 对象；如果基础地址不合法或不能追加路径，就返回输入错误。

调用关系：list_remote_control_clients 和 revoke_remote_control_client 都先调用它获得环境客户端地址。列表操作直接使用这个地址；撤销操作会在这个地址后面再追加 client_id。基础地址的规范化工作交给 normalize_remote_control_base_url。

调用图：调用 1 个内部函数（normalize_remote_control_base_url）；被 2 处调用（list_remote_control_clients, revoke_remote_control_client）。

fn try_from(client: RemoteControlClientResponse) -> Result<Self, Self::Error>

作用：这个转换函数把服务器返回的客户端原始数据，变成项目内部统一使用的 RemoteControlClient。它特别负责把 last_seen_at 这种时间字符串转成更方便计算的时间戳。

数据流：进去的是 RemoteControlClientResponse，里面是从服务器 JSON 解出来的字段，比如 client_id、设备名称、平台、系统版本和 last_seen_at。它把大多数字段原样搬到 RemoteControlClient；如果 last_seen_at 存在，就按 RFC3339 这种常见互联网时间格式解析，再转成 Unix 时间戳，也就是从 1970 年开始算的秒数。出来的是转换好的 RemoteControlClient；如果时间格式不对，就返回数据无效错误。

调用关系：list_remote_control_clients 在解析完服务器列表后，会对每个条目调用它。这样上层拿到的不是松散的服务器原始格式，而是项目协议里定义好的客户端对象；如果某个客户端时间字段坏了，错误会在列表转换时被明确报出来。

fn local(response: Response) -> Self

作用：把一个本地 HTTP 响应包装成可逐块读取的正文流。有人已经拿到了 reqwest 的 Response，但不想一次性读完正文时，会用它。

数据流：输入是一个本地 HTTP 响应对象 → 它取出响应里的字节流，并把这个流固定在内存位置上，方便异步读取 → 输出一个 HttpResponseBodyStream，之后调用 recv 就能一块一块拿到正文内容。

调用关系：它由 http_request_stream 在本地执行 HTTP 请求成功后调用。它不再把工作交给远端路由表，而是直接依赖 reqwest 提供的 bytes_stream 来读网络返回的数据。

调用图：被 1 处调用（http_request_stream）；外部调用 2 个（pin, bytes_stream）。

fn remote(
        inner: Arc<Inner>,
        request_id: String,
        rx: mpsc::Receiver<HttpRequestBodyDeltaNotification>,
    ) -> Self

作用：创建一个远端 HTTP 响应正文流。远端执行请求时，正文不是直接在本进程里，而是通过通知一块块送回来，这个函数负责准备接收这些通知。

数据流：输入是共享的客户端内部状态、这次请求的编号，以及接收正文分片的通道 → 它记录下下一个应该收到的分片序号，从 1 开始，并标记流还没关闭 → 输出一个 HttpResponseBodyStream，调用者可以像读普通流一样读远端传来的正文。

调用关系：它由 http_request_stream 在发起远端流式 HTTP 请求后调用。后续真正的数据会先进入 Inner::handle_http_body_delta_notification，再通过通道送到这个流的 recv。

调用图：被 1 处调用（http_request_stream）。

async fn recv(&mut self) -> Result<Option<Vec<u8>>, ExecServerError>

作用：读取下一块 HTTP 响应正文。它是使用者真正拿数据的入口：有数据就返回一段字节，到结尾就返回没有数据，出错就返回错误。

数据流：输入是当前流的内部状态，以及可能已经到达的网络字节或远端通知 → 本地模式直接等 reqwest 的下一块字节；远端模式从通道等下一条分片通知，检查序号是否连续，检查是否带错误或结束标记，并在结束时清理路由 → 输出是下一段正文 Vec<u8>、正文结束的 None，或者协议/HTTP 错误；同时可能把远端流标记为关闭并从路由表移除。

调用关系：它由 collect_body 这类需要收集响应正文的代码反复调用。远端模式下，它会在结束、通道关闭、序号不对或远端报错时调用 finish_remote_stream，把这次请求的收件路线拆掉。

调用图：调用 1 个内部函数（finish_remote_stream）；被 1 处调用（collect_body）；外部调用 3 个（HttpRequest, Protocol, format!）。

fn drop(&mut self)

作用：在正文流被直接丢弃时做善后。也就是说，调用者还没读到结尾就不读了，它会安排清理远端的接收路线。

数据流：输入是即将被释放的 HttpResponseBodyStream → 如果它是远端流，并且还没有关闭，就把 closed 标记改成 true，然后安排一个异步任务去删除请求编号对应的路由 → 没有普通返回值，但会避免后续远端分片继续发到没人接的通道。

调用关系：这是 Rust 的 Drop 自动触发的清理动作，不需要外部显式调用。它把异步清理交给 spawn_remove_http_body_stream，因为 drop 本身不能直接 await 等异步操作完成。

调用图：调用 1 个内部函数（spawn_remove_http_body_stream）；外部调用 1 个（clone）。

fn new(inner: Arc<Inner>, request_id: String) -> Self

作用：为一个即将开始的远端 HTTP 正文流创建“临时登记”。它的作用是：请求还在路上时，先保证这个请求编号有清理保障。

数据流：输入是共享内部状态和请求编号 → 它保存这两样信息，并把 active 设为 true，表示如果对象被丢弃就需要自动清理 → 输出一个 HttpBodyStreamRegistration。

调用关系：它由 http_request_stream 在发起流式 HTTP 请求前后使用。这个登记对象配合 HttpBodyStreamRegistration::drop，保证请求中途取消时也不会留下废弃路由。

调用图：被 1 处调用（http_request_stream）。

fn disarm(&mut self)

作用：解除临时登记的自动清理。意思是：现在正式的正文流已经接管了清理责任，这个临时保险不用再生效。

数据流：输入是一个登记对象本身 → 它把 active 从 true 改成 false → 没有返回值；之后这个登记对象被丢弃时，不会再删除路由。

调用关系：它通常在请求已经顺利拿到响应并创建了真正的 HttpResponseBodyStream 后使用。这样可以避免临时登记和正式流重复清理同一条路线。

fn drop(&mut self)

作用：在临时登记被丢弃时自动清理路线。它主要防止“请求还没拿到响应头就被取消”时，路由表里留下没人用的请求编号。

数据流：输入是即将释放的登记对象 → 如果 active 仍然为 true，就复制内部状态和请求编号，并安排异步删除路由 → 没有普通返回值，但会减少资源泄漏和错误投递。

调用关系：这是自动触发的 Drop 清理逻辑。它和 HttpBodyStreamRegistration::disarm 配合：没 disarm 说明流程没正常交接，就调用 spawn_remove_http_body_stream 做兜底清理。

调用图：调用 1 个内部函数（spawn_remove_http_body_stream）；外部调用 1 个（clone）。

async fn finish_remote_stream(inner: &Arc<Inner>, request_id: &str, closed: &mut bool)

作用：正式结束一个远端正文流，并从路由表里移除它。它保证同一条流只清理一次。

数据流：输入是共享内部状态、请求编号，以及 closed 标记 → 如果已经关闭就直接返回；否则先把 closed 改成 true，再调用内部状态删除这条请求编号的通道 → 没有返回值，但路由表会少掉这条流。

调用关系：它由 HttpResponseBodyStream::recv 在读到结尾、遇到错误、通道断开或序号异常时调用。它把具体删除动作交给 Inner::remove_http_body_stream。

调用图：被 1 处调用（recv）。

fn spawn_remove_http_body_stream(inner: Arc<Inner>, request_id: String)

作用：从不能等待异步操作的地方，安排一次“稍后删除正文流路由”的任务。主要给 Drop 清理使用。

数据流：输入是共享内部状态和请求编号 → 它尝试拿到当前 Tokio 运行时句柄；如果拿到了，就启动一个后台异步任务去删除对应路由 → 没有直接返回清理结果，清理由后台任务完成。

调用关系：它被 HttpResponseBodyStream::drop 和 HttpBodyStreamRegistration::drop 调用。因为 drop 函数不能直接执行 await，所以这里充当“把清理工作丢给异步运行时”的桥梁。

调用图：被 2 处调用（drop, drop）；外部调用 1 个（try_current）。

async fn send_body_delta(
    notifications: &RpcNotificationSender,
    delta: HttpRequestBodyDeltaNotification,
) -> bool

作用：把一块 HTTP 响应正文分片作为通知发出去。远端需要把正文一段段传给对方时，会用它。

数据流：输入是通知发送器和一条正文分片通知 → 它用固定的方法名发送这条通知 → 输出 true 或 false，表示通知是否成功发出。

调用关系：它位于远端发送正文分片的出口处。接收端会通过 Inner::handle_http_body_delta_notification 处理同类通知，再送到对应的 HttpResponseBodyStream。

调用图：调用 1 个内部函数（notify）。

async fn handle_http_body_delta_notification(
        &self,
        params: Option<Value>,
    ) -> Result<(), ExecServerError>

作用：接收一条远端发来的 HTTP 正文分片通知，并把它送进正确请求的通道。它就像分拣员，根据 request_id 找收件箱。

数据流：输入是可能为空的 JSON 参数 → 它先把 JSON 转成 HttpRequestBodyDeltaNotification；再按 request_id 查路由表，找到对应发送端后尝试投递；如果这是结束或错误分片，就移除路由；如果通道关闭或已满，就记录或清理错误 → 输出成功或解析错误，同时可能改动路由表和失败记录表。

调用关系：它由 RPC 通知处理流程在收到 http/request/bodyDelta 通知时调用。它会调用 remove_http_body_stream 做清理，必要时调用 record_http_body_stream_failure 记录“没能送达”的失败原因。

调用图：调用 2 个内部函数（record_http_body_stream_failure, remove_http_body_stream）；外部调用 2 个（debug!, from_value）。

async fn fail_all_http_body_streams(&self, message: String)

作用：让所有还在等待正文的 HTTP 流立刻失败。比如连接断了，如果不这么做，调用者可能会一直傻等。

数据流：输入是一段错误说明 → 它锁住路由表，取出当前所有流并清空路由表；然后给每个流尝试发送一条带错误的结束通知；如果发送失败，就把失败原因记到失败表里 → 没有返回值，但所有活动正文流都会被唤醒或留下可查询的错误。

调用关系：它通常在传输连接断开、通知处理失败等全局故障时被上层调用。它和 HttpResponseBodyStream::recv 配合，保证等待正文的人能收到错误，而不是无限等待。

调用图：外部调用 3 个（new, new, new）。

fn next_http_body_stream_request_id(&self) -> String

作用：生成一个新的 HTTP 正文流请求编号。每条远端正文流都需要唯一编号，才能把分片送回正确的人。

数据流：输入是内部计数器当前状态 → 它把计数器加一，并把数字格式化成类似 http-数字 的字符串 → 输出这个新的 request_id。

调用关系：它在准备注册新的远端流式 HTTP 请求时使用。生成出的编号随后会传给 insert_http_body_stream 和远端请求流程。

调用图：外部调用 1 个（format!）。

async fn insert_http_body_stream(
        &self,
        request_id: String,
        tx: mpsc::Sender<HttpRequestBodyDeltaNotification>,
    ) -> Result<(), ExecServerError>

作用：把一个请求编号和它的正文接收通道登记到路由表里。没有这一步，远端发回来的正文分片就不知道该送给谁。

数据流：输入是请求编号和一个通道发送端 → 它先加锁，检查这个编号是否已经存在；如果重复就返回协议错误；否则复制当前路由表，加入新条目并替换旧表；同时清掉同编号的旧失败记录 → 输出成功或错误，并改变内部路由表。

调用关系：它在发起远端流式 HTTP 请求前由上层流程调用。之后 Inner::handle_http_body_delta_notification 会依靠这里登记的通道来投递分片。

调用图：外部调用 3 个（new, Protocol, format!）。

async fn remove_http_body_stream(
        &self,
        request_id: &str,
    ) -> Option<mpsc::Sender<HttpRequestBodyDeltaNotification>>

作用：从路由表里删除一个 HTTP 正文流。正文结束、出错、调用者放弃读取时，都需要它来收尾。

数据流：输入是请求编号 → 它加锁读取当前路由表；如果找不到就返回 None；如果找到了，就复制路由表、移除该编号、替换旧表 → 输出被移除的通道发送端，或者 None；内部路由表会被更新。

调用关系：它被 Inner::handle_http_body_delta_notification 在终止分片或通道异常时调用，也会被 finish_remote_stream 间接用于 recv 的收尾流程。它是清理远端正文路线的核心小工具。

调用图：被 1 处调用（handle_http_body_delta_notification）；外部调用 1 个（new）。

async fn record_http_body_stream_failure(&self, request_id: &str, message: String)

作用：记录某个 HTTP 正文流的失败原因。特别是在分片没法送进通道时，它保存错误，方便后面读取者知道发生了什么。

数据流：输入是请求编号和错误文字 → 它加锁读取失败表，复制一份，写入这条请求的错误，再替换回内部状态 → 没有返回值，但失败表会多一条记录或更新一条记录。

调用关系：它由 Inner::handle_http_body_delta_notification 在通道已满、分片无法投递时调用。之后 HttpResponseBodyStream::recv 在发现通道关闭时，可以通过 take_http_body_stream_failure 取出这个错误。

调用图：被 1 处调用（handle_http_body_delta_notification）；外部调用 1 个（new）。

async fn take_http_body_stream_failure(&self, request_id: &str) -> Option<String>

作用：取出并删除某个 HTTP 正文流之前记录的失败原因。它像“读一次就撕掉”的错误便签。

数据流：输入是请求编号 → 它加锁查看失败表；如果有对应错误，就复制出来，并从失败表中移除；如果没有就返回 None → 输出错误文字或 None，同时可能清掉内部失败记录。

调用关系：它由 HttpResponseBodyStream::recv 在远端通道关闭后使用，用来判断这是正常结束，还是之前有投递失败等异常。它和 record_http_body_stream_failure 成对工作。

调用图：外部调用 1 个（new）。

async fn open(
    client: ExecServerClient,
    path: PathUri,
    sandbox: Option<FileSystemSandboxContext>,
) -> FileSystemResult<FileSystemReadStream>

作用：这个函数开始一次远程文件读取，并返回一个可以一块一块吐出文件内容的读取流。调用者用它时，不需要关心远端文件是怎么打开、怎么分块、怎么关闭的。

数据流：进去的是远端执行服务的客户端 client、要读的路径 path，以及可选的沙箱信息 sandbox（沙箱就是限制文件访问范围的安全边界）。函数先生成一个唯一的 handle_id，记录当前异步运行环境，然后把 path 和 sandbox 发给远端打开文件。打开成功后，它返回一个 FileSystemReadStream；这个流之后每次被读取，都会向远端请求从某个 offset 开始、最多 FILE_READ_CHUNK_SIZE 大小的一块数据。读到结尾时会关闭远端句柄；如果远端返回异常数据，就把它变成读取错误。

调用关系：它是外部代码进入“远程读文件”流程的入口。它内部会创建 FileReadRegistration，当成这次远程打开文件的登记单；还会用 try_unfold 搭出一个按需取数据的流。每次流需要下一块内容时，它把活儿交给 client.fs_read_block；最后读完时交给 client.fs_close 收尾。如果中途出了远端错误，会通过 map_remote_error 转成文件系统层能理解的错误。

调用图：调用 1 个内部函数（new）；外部调用 3 个（new_v4, try_unfold, try_current）。

fn drop(&mut self)

作用：这个函数是安全兜底：当一次远程读文件的登记对象被销毁时，如果文件还没正常关闭，它会尽量通知远端关闭。这样可以减少远端文件句柄泄漏，就像借了钥匙没还时系统自动帮你归还。

数据流：进去的是即将被丢弃的 FileReadRegistration，里面有 client、handle_id、运行时句柄 runtime，以及 active 标记。它先看 active：如果已经关闭，就什么也不做；如果还活跃，就复制 client 和 handle_id，找一个可用的 tokio 异步运行时（一种负责跑异步任务的调度器），然后派发一个后台任务去调用远端 fs_close。它不返回业务结果，也不报关闭失败，因为这是对象销毁时的补救动作。

调用关系：它在 open 创建的读取流生命周期结束时自动触发，不需要调用者手动调用。正常读到文件结尾时，open 里的流逻辑会先调用 fs_close 并把 active 设为 false，这样 drop 就不会重复关闭。只有读取被提前取消、流被丢弃等情况，drop 才接过收尾工作，把关闭请求交给异步运行时后台执行。

调用图：外部调用 1 个（clone）。

fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result

作用：这个函数控制公钥对象在调试日志里怎么显示。它会显示协议套件名字，但把真正的公钥内容遮住，避免日志里泄露敏感材料。

数据流：进去的是一个公钥对象和日志格式化工具 → 它把对象包装成可读的调试结构，同时把两段公钥写成“已隐藏” → 出来的是一段安全的调试文本，不改动原对象。

调用关系：它通常不是业务流程主动调用的，而是在打印 Debug 信息时由 Rust 自动用到。它把具体格式化工作交给标准的 debug_struct 工具。

调用图：外部调用 1 个（debug_struct）。

fn decode(&self) -> Result<(<X25519 as Dh>::PubKey, MlKem768PublicKey), NoiseChannelError>

作用：这个函数把注册表或配置里保存的公钥文字还原成加密库能用的真实公钥。它还会检查协议套件和长度，防止拿错协议的钥匙来用。

数据流：进去的是带 suite、X25519 公钥、ML-KEM-768 公钥的对象，其中公钥是 base64 字符串 → 它先确认 suite 是本文件支持的那一种，再解码并检查长度 → 出来的是两段加密库可直接使用的公钥；如果格式不对，就返回明确的公钥错误。

调用关系：InitiatorHandshake::start 在发起握手前会调用它，因为发起方必须先确认自己要连接的 exec-server 公钥是合法且匹配的。它内部会在 ML-KEM 公钥上使用 from_slice，并在失败时生成 InvalidPublicKey 错误。

调用图：被 1 处调用（start）；外部调用 2 个（from_slice, InvalidPublicKey）。

fn generate() -> Result<Self, NoiseChannelError>

作用：这个函数生成一套本端长期使用的 Noise 身份钥匙。没有它，进程就没有办法向对方证明“我是我”。

数据流：进去不需要业务输入 → 它分别生成 X25519 密钥对和 ML-KEM-768 密钥对，前者是传统椭圆曲线密钥交换，后者是抗量子密钥封装算法 → 出来的是一个 NoiseChannelIdentity；如果随机生成失败，就返回密钥生成错误。

调用关系：创建环境、测试握手、harness 连接等流程都会用它准备身份。它把真正的随机密钥生成交给 X25519::genkey 和 MlKem768::genkey。

调用图：被 22 处调用（upsert_noise_environment, hybrid_ik_roundtrip_authenticates_both_endpoints, initiator_rejects_oversized_handshake_payload, initiator_rejects_wrong_responder_key, public_key_serializes_with_expected_suite, public_key_validation_rejects_unknown_suite, responder_rejects_mismatched_prologue, transport_rejects_replayed_ciphertext, transport_rejects_tampered_ciphertext, processor_exit_reports_closed_virtual_stream (+12 more)）；外部调用 2 个（genkey, genkey）。

fn public_key(&self) -> NoiseChannelPublicKey

作用：这个函数从本端身份里取出可以公开给别人看的公钥。别人需要这份公钥，才能在握手时验证并加密给本端。

数据流：进去的是本端保存的私钥加公钥身份 → 它只拿出公钥部分，把两段公钥编码成 base64 文本，并附上固定的协议套件名 → 出来的是可序列化、可放进注册表或消息里的 NoiseChannelPublicKey，不暴露私钥。

调用关系：它常和 NoiseChannelIdentity::generate 搭配使用：先生成身份，再把公钥发布出去。发起方之后会把这份公钥交给 NoiseChannelPublicKey::decode 来检查和使用。

fn start(
        identity: &NoiseChannelIdentity,
        responder_public_key: &NoiseChannelPublicKey,
        prologue: &[u8],
        payload: &[u8],
    ) -> Result<(Self, Vec<u8>), NoiseChannelE

作用：这个函数由 harness 一侧调用，用来开始第一次握手。它会固定预期的 exec-server 公钥，并把短期授权信息塞进第一条加密握手消息里。

数据流：进去的是 harness 自己的身份、exec-server 公钥、prologue 上下文、以及要随握手发送的 payload → 它先解码并检查对方公钥，再创建 Noise hybrid IK 握手，检查 payload 不超过协议最大长度，然后写出第一条握手消息 → 出来的是一个还没完成的 InitiatorHandshake 状态，以及要发给 exec-server 的字节消息。

调用关系：它是 harness 连接流程的起点，noise_harness_connection_from_websocket 和多种测试都会调用它。它会先调用 NoiseChannelPublicKey::decode，再把参数交给 clatter 库创建和写入握手消息；后续必须调用 InitiatorHandshake::finish 才能进入传输模式。

调用图：调用 1 个内部函数（decode）；被 13 处调用（hybrid_ik_roundtrip_authenticates_both_endpoints, initiator_rejects_oversized_handshake_payload, initiator_rejects_wrong_responder_key, responder_rejects_mismatched_prologue, transport_rejects_replayed_ciphertext, transport_rejects_tampered_ciphertext, processor_exit_reports_closed_virtual_stream, noise_harness_connection_from_websocket, duplicate_handshakes_exhaust_failure_budget, oversized_harness_authorization_is_rejected_before_validation (+3 more)）；外部调用 4 个（new, new, noise_hybrid_ik, InvalidMessage）。

fn finish(mut self, response: &[u8]) -> Result<NoiseTransport, NoiseChannelError>

作用：这个函数处理 exec-server 回来的第二条握手消息，并把发起方切换到真正的加密传输状态。它要求对方这条响应不能夹带应用数据。

数据流：进去的是尚未完成的发起方握手状态和 exec-server 的响应字节 → 它先检查响应长度，再解开握手消息，确认里面没有额外 payload，最后完成握手 → 出来的是 NoiseTransport，也就是后续可加密收发记录的通道；原来的握手状态被消耗掉，不能重复使用。

调用关系：它接在 InitiatorHandshake::start 之后使用。它调用 ensure_noise_frame_len 做长度保护，再交给 clatter 的 read_message 和 finalize 完成协议收尾。

调用图：调用 1 个内部函数（ensure_noise_frame_len）；外部调用 3 个（finalize, read_message, InvalidMessage）。

fn read_request(
        identity: &NoiseChannelIdentity,
        prologue: &[u8],
        request: &[u8],
    ) -> Result<Self, NoiseChannelError>

作用：这个函数由 exec-server 一侧调用，用来读取 harness 发来的第一条握手消息。它会从已认证的握手里取出 harness 的公钥，但还不会直接放行连接。

数据流：进去的是 exec-server 自己的身份、prologue 上下文、以及 harness 的第一条握手消息 → 它检查消息长度，创建响应方握手，读取并解密 payload，然后从握手状态里取出对方已认证的静态公钥 → 出来的是 PendingResponderHandshake，里面带着待授权的 harness 公钥和第一条消息里的 payload。

调用关系：run_multiplexed_environment 和相关测试会在收到新连接请求时调用它。它完成的是“认出对方是谁”，但授权判断要由调用者去注册表做；通过后才会调用 PendingResponderHandshake::complete。

调用图：调用 1 个内部函数（ensure_noise_frame_len）；被 5 处调用（hybrid_ik_roundtrip_authenticates_both_endpoints, transport_rejects_replayed_ciphertext, transport_rejects_tampered_ciphertext, processor_exit_reports_closed_virtual_stream, run_multiplexed_environment）；外部调用 4 个（new, new, noise_hybrid_ik, InvalidMessage）。

fn complete(mut self) -> Result<(NoiseTransport, Vec<u8>), NoiseChannelError>

作用：这个函数在注册表确认 harness 公钥有权限后调用，用来完成 exec-server 侧握手。它会生成第二条握手响应，并进入加密传输模式。

数据流：进去的是已经读过请求、等待授权的握手状态 → 它写出一个空 payload 的响应消息，然后 finalize 完成协议 → 出来的是 NoiseTransport 和需要发回 harness 的响应字节；这个 pending 状态也被消耗掉，避免同一次握手被重复完成。

调用关系：它接在 PendingResponderHandshake::read_request 之后，但中间必须先经过授权检查。它把写响应和收尾工作交给 clatter 的 write_message 和 finalize。

调用图：外部调用 2 个（finalize, write_message）。

fn encrypt(&mut self, plaintext: &[u8]) -> Result<Vec<u8>, NoiseChannelError>

作用：这个函数把下一条明文记录加密成可以发送的密文记录。它还会检查长度，避免生成超过 Noise 协议允许大小的帧。

数据流：进去的是一段要发送的明文 bytes → 它加上 AES-GCM 认证标签需要的长度，确认总长度不超限，然后用当前发送计数器加密 → 出来的是密文字节；同时通道内部的发送状态会前进一步，所以同一条逻辑消息不应该重复加密重试。

调用关系：spawn_noise_virtual_stream 在把虚拟流数据发出去前会调用它。它依赖 ensure_noise_frame_len 做大小检查，再交给 transport.send_vec 进行真正加密。

调用图：调用 1 个内部函数（ensure_noise_frame_len）；被 1 处调用（spawn_noise_virtual_stream）；外部调用 3 个（tag_len, send_vec, InvalidMessage）。

fn decrypt(&mut self, ciphertext: &[u8]) -> Result<Vec<u8>, NoiseChannelError>

作用：这个函数把收到的下一条密文记录解开成明文。它要求记录按顺序到达，因为加密通道内部有隐含的接收计数器。

数据流：进去的是一段收到的密文字节 → 它先确认长度至少能放下 AES-GCM 的防篡改标签，再检查不超过最大帧长度，最后用当前接收状态解密和验真 → 出来的是明文字节；如果消息被篡改、重放或顺序不对，就会失败。

调用关系：receive_data 在收到远端数据时会调用它。它先做本地长度防护，再把真正的解密交给 transport.receive_vec。

调用图：调用 1 个内部函数（ensure_noise_frame_len）；被 1 处调用（receive_data）；外部调用 3 个（tag_len, receive_vec, InvalidMessage）。

fn noise_channel_prologue(
    environment_id: &str,
    executor_registration_id: &str,
    stream_id: &str,
) -> Vec<u8>

作用：这个函数生成握手前双方共同绑定的一段上下文。它把环境、执行器注册、具体流这三件事绑到握手里，防止消息被搬到别的连接里冒用。

数据流：进去的是 environment_id、executor_registration_id、stream_id 三个字符串 → 它先放入固定的协议域名，再依次把三个 ID 以“长度加内容”的形式拼进去 → 出来的是一段 prologue 字节，双方必须用同一段才能握手成功。

调用关系：harness 建连和 exec-server 接收连接时都会调用它，测试里也用它制造匹配或不匹配的场景。它把每一段拼接工作交给 append_prologue_part。

调用图：调用 1 个内部函数（append_prologue_part）；被 7 处调用（noise_harness_connection_from_websocket, run_multiplexed_environment, duplicate_handshakes_exhaust_failure_budget, oversized_harness_authorization_is_rejected_before_validation, pending_harness_key_validation_does_not_block_new_handshakes, repeated_early_data_during_validation_closes_the_physical_relay, repeated_malformed_handshakes_close_the_physical_relay）；外部调用 1 个（new）。

fn append_prologue_part(prologue: &mut Vec<u8>, part: &[u8])

作用：这个小函数把 prologue 的一小段安全地追加进去。它会先写长度，再写内容，避免不同字段拼在一起后看起来一样。

数据流：进去的是正在构建的 prologue 缓冲区和一段要追加的字节 → 它把这段内容的长度转成 8 字节大端格式写入，再写入内容本身 → 出来是被扩展后的 prologue 缓冲区。

调用关系：它只被 noise_channel_prologue 调用，是那个函数的拼装零件。这样做比直接字符串相加安全，因为字段边界不会混淆。

调用图：被 1 处调用（noise_channel_prologue）。

fn ensure_noise_frame_len(
    frame_len: usize,
    message: &'static str,
) -> Result<(), NoiseChannelError>

作用：这个函数统一检查 Noise 消息帧有没有超过协议最大长度。它像门口的限高杆，太大的消息直接挡掉。

数据流：进去的是某个帧的长度和出错时要显示的说明文字 → 它把长度和 MAX_MESSAGE_LEN 比较 → 如果合格就什么也不改并返回成功；如果太大就返回 InvalidMessage 错误。

调用关系：握手完成、读取请求、加密、解密这些入口都会调用它。它让所有地方使用同一条大小规则，避免某个路径漏掉长度检查。

调用图：被 4 处调用（finish, decrypt, encrypt, read_request）；外部调用 1 个（InvalidMessage）。

fn from(error: clatter::error::TransportError) -> Self

作用：这个函数把底层加密库的错误转换成本文件统一使用的 NoiseChannelError。这样上层不用认识很多种底层错误类型。

数据流：进去的是 clatter 库返回的握手错误或传输错误 → 它把错误转成字符串，并包进 Handshake 或 Transport 这类更贴近本文件语境的错误 → 出来的是统一的 NoiseChannelError，不改动其他状态。

调用关系：本文件里很多地方使用问号操作符返回错误时会自动用到它。它把底层 clatter 的失败翻译成上层调用者能统一处理的错误。

调用图：外部调用 4 个（Handshake, Transport, to_string, to_string）。

fn push(
        &mut self,
        seq: u32,
        payload: Vec<u8>,
    ) -> Result<Vec<Vec<u8>>, ExecServerError>

作用：接收一条带序号的密文，把它放到正确位置；如果从当前该读的序号开始已经连成一段，就把这一段按顺序交出去。调用它的人不用自己处理乱序、重复和缓存爆满这些麻烦事。

数据流：进去的是一个序号 seq 和一段密文 payload，同时它会查看自己记着的 next_seq、已暂存的 pending、以及暂存字节数 pending_bytes。它先判断这条是不是旧消息或重复消息，是的话直接返回空列表；如果这条太超前，就检查是否超过允许窗口或缓存上限，安全的话先存起来；如果正好是下一条该收的消息，就把它和后面已经缓存且连续的消息依次取出。出来的是一组已经排好顺序、可以立刻交给解密层的密文；过程中可能更新 next_seq、pending 和 pending_bytes，也可能返回协议错误。

调用关系：它在 receive_data 收到中继数据时被调用，是密文真正进入 Noise 解密前的一道排队关卡。遇到可以放行的连续消息时，它会反复调用 OrderedCiphertextFrames::advance 推进“下一条该收的编号”；遇到异常情况时，它会构造协议错误，把问题往上交给接收流程处理。

调用图：调用 1 个内部函数（advance）；被 2 处调用（receive_data, receive_data）；外部调用 3 个（new, Protocol, vec!）。

fn advance(&mut self) -> Result<(), ExecServerError>

作用：把“下一条该收的序号”往后挪一位。它是一个小而关键的保险步骤，确保序号不会悄悄溢出变回 0。

数据流：进去的是当前对象里的 next_seq。它尝试把这个数字加 1；如果数字还在 u32 能表示的范围内，就保存新的 next_seq 并返回成功；如果已经加不了了，就返回一个协议错误，说明序号用尽。它不处理密文本身，只改这个顺序指针。

调用关系：它只被 OrderedCiphertextFrames::push 使用。push 每成功放行一条密文，就靠它把期待的序号往前推进，这样下一次才能判断新来的消息是正好该收、太早、重复，还是太超前。

调用图：被 1 处调用（push）。

fn frame_jsonrpc_message(message: &JSONRPCMessage) -> Result<Vec<u8>, ExecServerError>

作用：把一条 JSON-RPC 消息变成适合放进加密字节流里的格式。它会在消息前面加上 4 个字节的长度，方便接收方以后知道该从哪里切开。

数据流：进去的是一条 JSONRPCMessage。函数先准备 4 个空字节占位，再用 JSON 序列化把消息写到后面；接着算出真正消息内容有多长，如果超过上限就返回协议错误；如果没问题，就把长度按固定格式写回最前面。出来的是一整段字节：前 4 字节是长度，后面是 JSON 内容。

调用关系：它在发送前使用。spawn_noise_virtual_stream 需要开出虚拟加密流时会用它把消息打包；processor_exit_reports_closed_virtual_stream 在报告处理器退出并关闭虚拟流时也会用它。它自己把 JSON 写入工作交给 serde_json::to_writer，遇到协议问题则生成 ExecServerError::Protocol。

调用图：被 2 处调用（spawn_noise_virtual_stream, processor_exit_reports_closed_virtual_stream）；外部调用 3 个（Protocol, to_writer, vec!）。

fn push(
        &mut self,
        plaintext_record: &[u8],
    ) -> Result<Vec<JSONRPCMessage>, ExecServerError>

作用：把收到的一小段解密数据放进缓存里，并尽可能从里面拼出完整的 JSON-RPC 消息。它适合处理“消息被拆成多段”或“一段里含多条消息”的情况。

数据流：进去的是一段已经解密的明文字节。函数先检查这段记录是否太大，然后追加到内部 buffered 缓存；只要缓存里至少有 4 字节长度前缀，就读出声明的消息长度，检查长度是否合法；如果完整消息还没到齐，就先等下一段；如果已经到齐，就把对应字节解析成 JSONRPCMessage，并从缓存里删掉已消费的数据。出来的是这次新拼好的消息列表，同时内部缓存会留下尚未拼完的尾巴。

调用关系：它在接收数据时使用，receive_data 收到解密后的记录后会调用它。它不直接管网络，也不直接管加密，只负责“把碎片拼成完整消息”。解析 JSON 的工作交给 serde_json::from_slice，读取长度时使用固定的 4 字节大端格式，也就是最高位字节放在前面的数字写法。

调用图：被 2 处调用（receive_data, receive_data）；外部调用 4 个（new, Protocol, from_slice, from_be_bytes）。

fn data(stream_id: String, seq: u32, payload: Vec<u8>) -> Self

作用：创建一个“数据帧”，也就是把一段真正要传的内容包装成中继协议能发送的小包。别人需要发送 JSON-RPC 内容时会先用它打包。

数据流：进去的是流编号、序号和原始字节内容 → 它把这些放进 RelayData，并补上协议版本、确认字段等固定信息 → 出来的是一个可编码发送的 RelayMessageFrame。

调用关系：它是发数据时的包装工；harness_connection_from_websocket 会用它把内部 JSON-RPC 消息包成 WebSocket 二进制帧，测试里也用它模拟远端发来的数据。

调用图：外部调用 1 个（Data）。

fn resume(stream_id: String) -> Self

作用：创建一个“恢复帧”，用来告诉对方某个流要开始或恢复通信。它像见面时先报一个通道号。

数据流：进去的是流编号 → 它生成带有 Resume 内容的中继帧，next_seq 从 0 开始 → 出来的是一个恢复用的 RelayMessageFrame。

调用关系：harness_connection_from_websocket 一建立 WebSocket 就先发送这个帧，让另一边知道后续数据属于哪个流。

调用图：外部调用 1 个（Resume）。

fn handshake(stream_id: String, payload: Vec<u8>) -> Self

作用：创建一个“握手帧”，用来传递加密握手阶段需要的字节内容。没有它，Noise 安全通道就没法互相确认身份。

数据流：进去的是流编号和握手字节 → 它把握手字节放进 RelayHandshake → 出来的是一个握手用的 RelayMessageFrame。

调用关系：run_multiplexed_environment 在授权通过并完成 Noise 握手后，会用它把握手响应发回对方。

调用图：外部调用 1 个（Handshake）。

fn reset(stream_id: String, reason: String) -> Self

作用：创建一个“重置帧”，表示某条流要被关闭或拒绝。它相当于告诉对方：这条线别再用了。

数据流：进去的是流编号和关闭原因 → 它把原因放进 RelayReset → 出来的是一个可发送的重置帧。

调用关系：send_reset 会调用它；run_multiplexed_environment 在握手失败、流不存在、资源过多等情况下通过它通知远端。

调用图：外部调用 1 个（Reset）。

fn validate(&self) -> Result<RelayFrameBodyKind, ExecServerError>

作用：检查一个中继帧是不是基本合规，并判断它是哪种类型。这样后面的代码不会拿坏包当正常消息处理。

数据流：进去的是一个 RelayMessageFrame → 它检查版本号、流编号、正文是否存在，以及数据帧/握手帧/重置帧的必要字段 → 出来的是帧类型，或者一个协议错误。

调用关系：它是所有入站帧的门卫；into_data 和 into_handshake_payload 会先调用它，harness_connection_from_websocket 和 run_multiplexed_environment 也依赖它决定下一步怎么处理。

调用图：被 2 处调用（into_data, into_handshake_payload）；外部调用 2 个（Protocol, format!）。

fn into_data(self) -> Result<RelayData, ExecServerError>

作用：把一个确认是数据类型的中继帧取出成 RelayData。它避免调用方手动拆包时误拿错类型。

数据流：进去的是一个中继帧 → 它先调用 validate 确认这是 Data 帧 → 出来的是里面的 RelayData；如果不是数据帧，就返回协议错误。

调用关系：into_jsonrpc_message 会接着用它取出字节内容；run_multiplexed_environment 收到数据帧后也靠它把内容交给虚拟流。

调用图：调用 1 个内部函数（validate）；被 1 处调用（into_jsonrpc_message）；外部调用 1 个（Protocol）。

fn into_jsonrpc_message(self) -> Result<JSONRPCMessage, ExecServerError>

作用：把数据帧里的字节内容解析成 JSON-RPC 消息。它完成“中继协议小包”到“业务请求/响应”的转换。

数据流：进去的是一个数据帧 → 它先取出 RelayData 的 payload，再按 JSON 格式反序列化 → 出来的是 JSONRPCMessage，或者 JSON 解析错误。

调用关系：harness_connection_from_websocket 收到远端数据时会用它，然后把解析出的 JSON-RPC 消息交给内部连接事件队列。

调用图：调用 1 个内部函数（into_data）；外部调用 1 个（from_slice）。

fn into_handshake_payload(self) -> Result<Vec<u8>, ExecServerError>

作用：从握手帧里取出握手用的原始字节。它保证只有真正的 Handshake 帧才能被当作加密握手数据使用。

数据流：进去的是一个中继帧 → 它先验证类型必须是 Handshake → 出来的是握手 payload；类型不对或字段缺失就返回协议错误。

调用关系：run_multiplexed_environment 收到新虚拟流的握手请求时会调用它，再把字节交给 PendingResponderHandshake 解析。

调用图：调用 1 个内部函数（validate）；外部调用 1 个（Protocol）。

fn into_reset_reason(self) -> Option<String>

作用：尝试从重置帧中拿到关闭原因。它只在原因非空时返回，避免传出没有意义的空字符串。

数据流：进去的是一个中继帧 → 它查看正文是否是 Reset 并且 reason 不为空 → 出来的是 Some(reason) 或 None。

调用关系：harness_connection_from_websocket 收到 Reset 时会用它，把断开原因包装成 JsonRpcConnectionEvent::Disconnected 发给内部使用者。

fn encode_relay_message_frame(frame: &RelayMessageFrame) -> Vec<u8>

作用：把中继帧编码成可以放进 WebSocket 二进制消息里的字节。网络上传不了 Rust 结构体，只能传字节。

数据流：进去的是 RelayMessageFrame 引用 → 它用 protobuf 编码把结构体压成字节数组 → 出来的是 Vec<u8>。

调用关系：所有发送中继帧的地方都会经过它，比如普通 WebSocket 连接、Noise 虚拟流和 send_reset；测试也用它造模拟数据。

调用图：被 10 处调用（spawn_noise_virtual_stream, noise_harness_connection_from_websocket, harness_connection_from_websocket, send_reset, harness_connection_sends_keepalive_and_receives_relay_data, duplicate_handshakes_exhaust_failure_budget, oversized_harness_authorization_is_rejected_before_validation, pending_harness_key_validation_does_not_block_new_handshakes, repeated_early_data_during_validation_closes_the_physical_relay, repeated_malformed_handshakes_close_the_physical_relay）；外部调用 1 个（encode_to_vec）。

fn decode_relay_message_frame(
    payload: &[u8],
) -> Result<RelayMessageFrame, ExecServerError>

作用：把 WebSocket 收到的二进制字节还原成中继帧。它是接收方向的入口检查之一。

数据流：进去的是一段字节 → 它按 protobuf 格式解码 → 出来的是 RelayMessageFrame；如果字节不是合法帧，就返回协议错误。

调用关系：harness_connection_from_websocket 和 run_multiplexed_environment 收到 Binary 消息后会先调用它；测试辅助函数也用它读取发出的帧。

调用图：被 5 处调用（noise_harness_connection_from_websocket, harness_connection_keeps_outbound_frame_while_send_is_backpressured, read_resume_stream_id, duplicate_handshakes_exhaust_failure_budget, oversized_harness_authorization_is_rejected_before_validation）；外部调用 1 个（decode）。

fn jsonrpc_payload(message: &JSONRPCMessage) -> Result<Vec<u8>, ExecServerError>

作用：把 JSON-RPC 消息转成字节，方便塞进中继数据帧。它做的是业务消息到网络载荷的转换。

数据流：进去的是 JSONRPCMessage → 它用 JSON 序列化变成字节数组 → 出来的是 Vec<u8>，或 JSON 序列化错误。

调用关系：harness_connection_from_websocket 发出内部 JSON-RPC 消息时会用同样思路；测试用它构造远端发来的数据帧。

调用图：被 1 处调用（harness_connection_sends_keepalive_and_receives_relay_data）；外部调用 1 个（to_vec）。

async fn send_event_with_keepalive(
    websocket: &mut T,
    keepalive: &mut tokio::time::Interval,
    incoming_tx: &mpsc::Sender<JsonRpcConnectionEvent>,
    event: JsonRpcConnectionEvent,
) -> Re

作用：把一个内部连接事件送进队列，同时在队列暂时塞满时继续给 WebSocket 发 Ping 保活。这样不会因为内部队列堵住就让外部连接看起来像死掉了。

数据流：进去的是 WebSocket、保活定时器、内部事件队列和要发送的事件 → 它一边等待队列有空间，一边按时间发送 Ping → 成功时事件进队列；如果队列关了或 WebSocket 写失败，就返回对应错误。

调用关系：harness_connection_from_websocket 在收到合法 JSON-RPC 消息后用它投递事件；测试 send_event_with_keepalive_pings_while_incoming_queue_is_full 专门验证队列满时仍会发保活。

调用图：被 1 处调用（send_event_with_keepalive_pings_while_incoming_queue_is_full）；外部调用 3 个（send, pin!, select!）。

fn harness_connection_from_websocket(
    stream: T,
    connection_label: String,
) -> JsonRpcConnection

作用：把一条普通 WebSocket 包装成项目内部统一使用的 JsonRpcConnection。调用方不用关心中继帧、Ping、断线和格式错误这些网络细节。

数据流：进去的是 WebSocket 流和一个连接标签 → 它生成流编号，建立进出两个消息队列，启动后台任务读写 WebSocket、编码/解码中继帧、发送保活 Ping、报告断线和坏消息 → 出来的是 JsonRpcConnection，内部代码通过它收发 JSON-RPC。

调用关系：这是普通 relay 连接的组装点；外层 connect_websocket 会调用它，内部又使用 encode_relay_message_frame、decode_relay_message_frame、jsonrpc_payload 和 send_event_with_keepalive。多个测试覆盖它的收发、断线、文本帧报错和写入背压行为。

调用图：调用 1 个内部函数（encode_relay_message_frame）；被 5 处调用（connect_websocket, harness_connection_keeps_outbound_frame_while_send_is_backpressured, harness_connection_reports_server_close, harness_connection_reports_text_frames_as_malformed, harness_connection_sends_keepalive_and_receives_relay_data）；外部调用 10 个（new_v4, resume, channel, select!, spawn, now, interval_at, vec!, channel, Binary）。

async fn run_multiplexed_environment(
    stream: WebSocketStream<S>,
    processor: ConnectionProcessor,
    environment_id: String,
    executor_registration_id: String,
    identity: NoiseChannelId

作用：在一条 executor WebSocket 上运行多条经过 Noise 认证的虚拟 JSON-RPC 流。它像一个总调度员：先验身份，再开通道，还要限制数量和清理坏连接。

数据流：进去的是 WebSocket、连接处理器、环境和注册编号、服务器的 Noise 身份，以及 harness 密钥验证器 → 它拆分读写端，启动物理写任务，循环读取中继帧；握手帧会先做 Noise 解析和授权校验，成功后创建 NoiseVirtualStream，数据帧转交给对应虚拟流，重置帧关闭对应流 → 结束时关闭所有活动流并停止写任务。

调用关系：run_remote_environment 会在需要服务远端环境时调用它；它把握手检查交给 PendingResponderHandshake 和 HarnessKeyValidator，把成功后的数据处理交给 spawn_noise_virtual_stream，并在异常时调用 send_reset 和 failed_handshake_budget_exhausted。

调用图：调用 4 个内部函数（read_request, noise_channel_prologue, failed_handshake_budget_exhausted, send_reset）；被 2 处调用（multiplexed_environment_sends_keepalive, run_remote_environment）；外部调用 16 个（new, new, from_utf8, clone, validate_harness_key, disconnect, receive_data, split, Protocol, take (+6 more)）。

fn failed_handshake_budget_exhausted(failed_handshakes: &mut usize) -> bool

作用：记录一次握手失败，并判断失败次数是否已经太多。它用一个小上限防止未授权对方反复触发昂贵的加密握手或注册表检查。

数据流：进去的是失败计数器的可修改引用 → 它把计数加一，再和最大允许失败次数比较 → 出来的是 true 或 false，表示是否该关闭整条物理连接。

调用关系：run_multiplexed_environment 在解析握手失败、重复握手、授权内容不合法等情况下调用它，用它决定只是拒绝一条流还是断开整个 relay。

调用图：被 1 处调用（run_multiplexed_environment）。

fn send_reset(physical_outgoing_tx: &mpsc::Sender<Vec<u8>>, stream_id: String)

作用：尽力给某条流排队发送一个重置帧，但不阻塞主循环。它用于快速通知对方“这条流被拒绝或关闭”。

数据流：进去的是物理发送队列和流编号 → 它创建 reset 帧，编码成字节，并尝试立即放入队列 → 没有显式返回结果；如果队列满或关闭，这次通知就放弃。

调用关系：run_multiplexed_environment 在很多拒绝场景都会调用它，例如握手失败、流数量超限、收到未知流数据；它内部使用 RelayMessageFrame::reset 和 encode_relay_message_frame。

调用图：调用 1 个内部函数（encode_relay_message_frame）；被 1 处调用（run_multiplexed_environment）；外部调用 2 个（try_send, reset）。

async fn harness_connection_sends_keepalive_and_receives_relay_data() -> anyhow::Result<()>

作用：测试普通 harness WebSocket 连接会发保活 Ping，并能把远端的数据帧转成内部 JSON-RPC 消息。它证明基础收包链路是通的。

数据流：进去的是测试创建的一对 WebSocket → 测试启动 harness_connection_from_websocket，读取初始 resume，等待 Ping，再发送一个编码后的数据帧 → 期望内部 incoming_rx 收到同一个 JSON-RPC 消息。

调用关系：它调用 websocket_pair、read_resume_stream_id、read_keepalive_ping、jsonrpc_payload 和 encode_relay_message_frame，围绕 harness_connection_from_websocket 做端到端验证。

调用图：调用 3 个内部函数（encode_relay_message_frame, harness_connection_from_websocket, jsonrpc_payload）；外部调用 8 个（assert!, data, read_keepalive_ping, read_resume_stream_id, test_jsonrpc_message, websocket_pair, Binary, Pong）。

async fn multiplexed_environment_sends_keepalive() -> anyhow::Result<()>

作用：测试 Noise 多路复用环境即使没有业务数据，也会给 WebSocket 发保活 Ping。这样可以及时发现连接是否还活着。

数据流：进去的是测试 WebSocket、运行路径、测试环境信息和允许所有密钥的验证器 → 它启动 run_multiplexed_environment → 期望另一端能读到 Ping，然后中止任务。

调用关系：它直接覆盖 run_multiplexed_environment 的物理写任务和保活逻辑，使用 AllowHarnessKeyValidator 避开授权复杂度。

调用图：调用 4 个内部函数（generate, run_multiplexed_environment, new, new）；外部调用 4 个（read_keepalive_ping, websocket_pair, current_exe, spawn）。

async fn validate_harness_key(
            &self,
            _harness_public_key: &NoiseChannelPublicKey,
            _authorization: &str,
        ) -> Result<(), ExecServerError>

作用：测试用的密钥验证器，永远说“允许”。它让测试可以专注于 relay 行为，而不是外部授权系统。

数据流：进去的是 harness 公钥和授权字符串，但测试实现不检查它们 → 直接返回 Ok → 不改动任何状态。

调用关系：multiplexed_environment_sends_keepalive 把它传给 run_multiplexed_environment；真实运行中会换成真正查询授权来源的实现。

async fn send_event_with_keepalive_pings_while_incoming_queue_is_full() -> anyhow::Result<()>

作用：测试内部事件队列满的时候，send_event_with_keepalive 不会傻等，而是继续发 Ping。它防止一个堵住的内部消费者拖死外部连接。

数据流：进去的是一个可控假 WebSocket 和容量很小的事件队列 → 测试先把队列塞满，再启动发送任务 → 期望先看到 Ping，清空队列后再看到目标 JSON-RPC 事件成功进入队列。

调用关系：它专门调用 send_event_with_keepalive，并用 ControlledWebSocket 模拟网络写入情况。

调用图：调用 1 个内部函数（send_event_with_keepalive）；外部调用 8 个（assert!, new, Message, test_jsonrpc_message, channel, spawn, now, interval_at）。

async fn harness_connection_reports_text_frames_as_malformed() -> anyhow::Result<()>

作用：测试普通 relay 收到文本 WebSocket 帧时会报告格式错误。这个协议只接受二进制 protobuf 帧，文本不该被悄悄忽略。

数据流：进去的是测试 WebSocket → 启动 harness_connection_from_websocket 后，服务端发送 Text 消息 → 期望内部收到 MalformedMessage，并说明需要二进制 protobuf 帧。

调用关系：它覆盖 harness_connection_from_websocket 的错误分支，依赖 websocket_pair 和 read_resume_stream_id 建好测试连接。

调用图：调用 1 个内部函数（harness_connection_from_websocket）；外部调用 4 个（assert!, read_resume_stream_id, websocket_pair, Text）。

async fn harness_connection_reports_server_close() -> anyhow::Result<()>

作用：测试远端关闭 WebSocket 时，内部连接能收到断开事件。这样上层不会一直以为连接还活着。

数据流：进去的是测试 WebSocket → 启动连接并读取初始 resume 后，服务端主动 close → 期望 incoming_rx 收到 Disconnected，且没有额外原因。

调用关系：它验证 harness_connection_from_websocket 对 Close 帧和连接结束的处理。

调用图：调用 1 个内部函数（harness_connection_from_websocket）；外部调用 3 个（assert!, read_resume_stream_id, websocket_pair）。

async fn harness_connection_keeps_outbound_frame_while_send_is_backpressured() -> anyhow::Result<()>

作用：测试 WebSocket 暂时写不出去时，待发送的业务消息不会丢。背压就是出口堵住了，代码应该等出口恢复再发。

数据流：进去的是可控假 WebSocket → 测试先阻塞写入，再往 outgoing_tx 放一个 JSON-RPC 消息，同时确认不会误报入站消息；恢复写入后 → 期望收到正确 stream_id 的数据帧，并能还原成原消息。

调用关系：它围绕 harness_connection_from_websocket 的发送循环，使用 ControlledWebSocketHandle 控制写入是否就绪，并调用 decode_relay_message_frame 检查结果。

调用图：调用 2 个内部函数（decode_relay_message_frame, harness_connection_from_websocket）；外部调用 8 个（from_secs, bail!, assert!, assert_eq!, new, test_jsonrpc_message, timeout, Pong）。

async fn websocket_pair() -> anyhow::Result<(
        WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>,
        WebSocketStream<tokio::net::TcpStream>,
    )>

作用：创建一对本机互连的 WebSocket，方便测试像真实网络一样收发消息。它相当于给测试搭了一个小型回环线路。

数据流：进去没有业务参数 → 它绑定本地 TCP 端口，启动服务端 accept，同时客户端 connect → 出来是一端客户端 WebSocket 和一端服务端 WebSocket。

调用关系：多个测试用它准备真实 tokio-tungstenite WebSocket，例如保活、文本帧错误和关闭事件测试。

调用图：外部调用 5 个（bind, format!, spawn, accept_async, connect_async）。

async fn read_resume_stream_id(
        websocket: &mut WebSocketStream<tokio::net::TcpStream>,
    ) -> anyhow::Result<String>

作用：测试辅助函数：从 WebSocket 读到第一条 resume 帧，并取出流编号。这样测试后续能用同一个 stream_id 发正确的数据帧。

数据流：进去的是服务端 WebSocket 可变引用 → 它限时读取下一条消息，要求是二进制，解码并验证为 Resume → 出来的是 stream_id。

调用关系：harness_connection_from_websocket 相关测试都用它确认连接启动时发出了 resume 帧，也顺便检查帧格式。

调用图：调用 1 个内部函数（decode_relay_message_frame）；外部调用 5 个（from_secs, next, bail!, assert_eq!, timeout）。

async fn read_keepalive_ping(
        websocket: &mut WebSocketStream<tokio::net::TcpStream>,
    ) -> anyhow::Result<()>

作用：测试辅助函数：一直读 WebSocket，直到看到 Ping。它用来证明保活机制真的在工作。

数据流：进去的是 WebSocket 可变引用 → 它在超时时间内循环读取消息，忽略普通数据和 Pong，遇到 Ping 就成功返回；如果连接先关闭或超时就报错。

调用关系：普通连接和多路复用连接的保活测试都会调用它。

调用图：外部调用 4 个（from_secs, next, bail!, timeout）。

fn test_jsonrpc_message() -> JSONRPCMessage

作用：生成一个简单固定的 JSON-RPC 请求，供测试反复使用。这样测试重点放在传输，而不是构造消息。

数据流：进去没有参数 → 它创建 id 为 1、method 为 test、没有参数的请求 → 出来的是 JSONRPCMessage::Request。

调用关系：多项测试用它作为标准样本，之后通过 jsonrpc_payload、RelayMessageFrame::data 或 outgoing_tx 走不同传输路径。

调用图：外部调用 2 个（Request, Integer）。

fn new(
            write_ready: bool,
        ) -> (
            Self,
            ControlledWebSocketHandle,
            futures_mpsc::UnboundedReceiver<Message>,
        )

作用：创建一个测试用的假 WebSocket，并返回控制手柄和出站消息接收器。它让测试能精确控制“网络现在能不能写”。

数据流：进去的是初始写入是否就绪 → 它创建入站/出站无界队列、几个原子状态和唤醒器 → 出来的是 ControlledWebSocket、本端控制手柄，以及用于观察发出消息的接收器。

调用关系：send_event_with_keepalive 和背压测试用它模拟真实 WebSocket，但不用真的开网络连接。

调用图：外部调用 5 个（clone, new, new, new, unbounded）。

fn send_inbound(&self, message: Message) -> anyhow::Result<()>

作用：往假 WebSocket 的入站方向塞一条消息，模拟远端发来了东西。

数据流：进去的是一条 WebSocket Message → 它把消息包装成 Ok 后放进入站队列 → 成功返回 Ok，失败则转成测试错误。

调用关系：背压测试用它在写入被堵住时发送 Pong，确认连接不会因为入站小消息而误丢待发业务帧。

调用图：外部调用 1 个（unbounded_send）。

fn set_write_blocked(&self)

作用：把假 WebSocket 设置成暂时不能写。它用来模拟网络出口堵塞。

数据流：进去没有额外数据 → 它把 write_ready 原子标志设为 false → 后续 poll_ready 会返回 Pending，而不是允许发送。

调用关系：harness_connection_keeps_outbound_frame_while_send_is_backpressured 用它制造背压场景。

fn set_write_ready(&self)

作用：把假 WebSocket 恢复成可以写，并唤醒等待写入的任务。它模拟网络恢复通畅。

数据流：进去没有额外数据 → 它把 write_ready 设为 true，并唤醒保存的 waker → 等待中的发送逻辑会继续执行。

调用关系：背压测试在确认写入已被阻塞后调用它，让 harness_connection_from_websocket 的发送循环继续把数据帧发出去。

async fn wait_for_blocked_write(&self) -> anyhow::Result<()>

作用：等待假 WebSocket 真的进入“写被堵住”的状态。这样测试不会因为时序太快而误判。

数据流：进去没有业务数据 → 它在最多一秒内轮询 write_blocked 标志，不满足时注册唤醒器 → 成功时返回 Ok，超时则测试失败。

调用关系：背压测试用它确认发送任务已经卡在 poll_ready，再继续发送入站消息和恢复写入。

调用图：外部调用 3 个（from_secs, poll_fn, timeout）。

fn poll_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>

作用：实现假 WebSocket 作为 Sink 时的“现在能不能写”检查。Sink 可以理解为异步写入口。

数据流：进去的是异步运行时给的上下文 → 如果 write_ready 为 true，就返回可以写；否则标记已阻塞、唤醒观察者并保存当前任务的唤醒器 → 出来是 Ready 或 Pending。

调用关系：futures 的发送流程会自动调用它；测试通过控制手柄改变 write_ready 来影响它的结果。

调用图：外部调用 2 个（waker, Ready）。

fn start_send(self: Pin<&mut Self>, item: Message) -> Result<(), Self::Error>

作用：实现假 WebSocket 的实际发送动作，把要发的消息放到测试可观察的出站队列里。

数据流：进去的是一条 Message → 它把消息放进 outbound_tx → 返回 Ok；测试随后从 outbound_rx 读到这条消息。

调用关系：send_event_with_keepalive 和 harness_connection_from_websocket 的发送代码会间接触发它，测试用出站接收器检查发出了 Ping、resume 或 data 帧。

调用图：外部调用 1 个（unbounded_send）。

fn poll_flush(
            self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
        ) -> Poll<Result<(), Self::Error>>

作用：实现假 WebSocket 的刷新动作，并且总是立即成功。测试里不需要模拟复杂的底层缓冲刷新。

数据流：进去的是异步上下文 → 它不做额外工作 → 直接返回 Ready(Ok)。

调用关系：SinkExt::send 在发送后可能需要 flush；这个实现让测试发送流程顺利完成。

调用图：外部调用 1 个（Ready）。

fn poll_close(
            self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
        ) -> Poll<Result<(), Self::Error>>

作用：实现假 WebSocket 的关闭动作，并且总是立即成功。它让测试对象满足 Sink 接口要求。

数据流：进去的是异步上下文 → 它不清理额外资源 → 直接返回 Ready(Ok)。

调用关系：虽然测试重点不是关闭流程，但 Sink 接口要求提供它，相关发送代码可以安全调用。

调用图：外部调用 1 个（Ready）。

fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>>

作用：实现假 WebSocket 作为 Stream 时的收消息动作。Stream 可以理解为异步读入口。

数据流：进去的是异步上下文 → 它从 inbound_rx 队列轮询下一条入站消息 → 出来是 Some(message)、None 或 Pending。

调用关系：harness_connection_from_websocket 的读取分支会间接调用它；测试通过 ControlledWebSocketHandle::send_inbound 往这个队列里喂消息。

调用图：外部调用 1 个（new）。