pub struct Environment { // 一个"执行环境" = 进程 + 文件系统 + HTTP，三套抽象打包在一起    remote_client: Option<LazyRemoteExecServerClient>, // 远程执行客户端；None 表示这是纯本地环境    ready_info: Arc<ArcSwapOption<EnvironmentReadyInfo>>, // 就绪信息用 ArcSwap 热替换：读方永不阻塞写方    provisioning_status_tx: Option<watch::Sender<Option<Result<(), String>>>>, // 供给进度广播通道（Pending / Ready / Failed）    startup_task: Arc<Mutex<Option<AbortOnDropHandle<()>>>>, // 后台启动任务句柄：Environment 被 drop 时自动取消，防泄漏    exec_backend: Arc<dyn ExecBackend>, // 统一执行后端 trait：LocalProcess 或 RemoteProcess 二选一    filesystem: Arc<dyn ExecutorFileSystem>, // 文件系统抽象（本地 / 远程 / 沙箱化三种实现）    http_client: Arc<dyn HttpClient>, // HTTP 出站通道，同样走这个环境——网络策略天然统一    local_runtime_paths: Option<ExecServerRuntimePaths>, // 本地模式下的运行时路径（shell、工具链位置）}

pub(crate) fn new( // 构造"懒"远程客户端：此刻不建任何连接，首次使用时才连    transport_params: ExecServerTransportParams, // 传输参数：WebSocket / Noise 会合 / Stdio 命令三种形态    http_client_factory: HttpClientFactory, // HTTP 工厂：流式响应与进程输出共用同一条连接的通告通道) -> Self {    Self {        transport_params: Some(transport_params), // 保存传输参数：连接时才知道走 WebSocket / Noise / Stdio        http_client_factory, // HTTP 工厂留待首次使用时创建客户端        recovery_policy: RecoveryPolicy::Wait, // 默认 Wait 策略：断线时等待重连，而不是立刻报错        startup: Arc::new(ConnectionAttempt::default()), // 首次连接槽位：所有调用者共享同一个结果（get_or_init 语义）        current_client: Arc::new(StdMutex::new(None)), // 最近一次成功的客户端缓存；每次重连成功后被替换        reconnect: Arc::new(StdMutex::new(None)), // 进行中的重连槽位：同一故障的多个调用者共享同一次尝试        refresh_lock: Arc::new(Mutex::new(())), // 串行化显式刷新操作，防止并发重建        environment_connection_state_tx: watch::channel(EnvironmentConnectionState::Disconnected).0, // 连接状态广播通道（初值 Disconnected）    }}

pub(crate) async fn get(&self) -> Result<ExecServerClient, ExecServerError> { // 取一个可用客户端：全 crate 的入口    if matches!(self.recovery_policy, RecoveryPolicy::FailFast) { // FailFast 模式（status 探测用）：只查缓存，绝不触发连接或重连        let client = match self.cached_client() { // FailFast 分支第一步：先看有没有现成缓存            Some(client) => client, // 有缓存直接用            None => match self.startup.result.get() { // 没有缓存：再看首次启动槽位的结果（成功/失败/未开始）                Some(Ok(client)) => client.clone(), // 首次启动已成功：复用那个结果                Some(Err(error)) => return Err(ExecServerError::ConnectionAttempt(Arc::clone(error))), // 首次启动已失败：原样上报，不重试                None => return Err(ExecServerError::Disconnected("exec-server environment is not ready".to_string())), // 还没人发起过连接：明确报"未就绪"            }, // 启动槽位三态（成功/失败/未开始）都处理完        }; // FailFast 分支的 client 已确定，交给下面统一返回        return client.fail_fast(); // 返回的客户端也继承 FailFast——后续 RPC 失败不触发恢复    }    if let Some(client) = self.connected_client() { // 快路径：已有连接且健康，直接返回（热路径零锁竞争）        return Ok(client); // 命中快路径，直接交付    }    let Some(cached_client) = self.cached_client() else { // 没有任何缓存客户端（首次使用）：走 initial_client        let client = self.initial_client().await?; // 首次调用：等 startup 槽位完成（所有并发者共享同一次连接）        if !client.is_disconnected() || !self.can_reconnect() { // 判断新客户端是否可以直接交付            return Ok(client); // 健康，或传输类型不支持重连（如 Stdio）：直接交付        }        return self.reconnect().await; // 断线且可重连：走合并重连    }; // else 分支结束：手里拿着一个已断线的缓存客户端    if !self.can_reconnect() { // 传输不支持重连（Stdio）：旧客户端原样交出去        return Ok(cached_client); // 不可重连的传输：把旧客户端交出去，让调用方自己决定怎么报错    }    self.reconnect().await // 常规路径：缓存已断线 → 合并重连后返回新客户端}

struct ActiveProcessStart { // RAII 计数器：这条连接上"正在启动中"的进程数    inner: Arc<Inner>, // 持有客户端内核：drop 时通过它减计数}impl Drop for ActiveProcessStart { // 实现 Drop：作用域退出即触发清理，不依赖调用纪律    fn drop(&mut self) { // 任何退出路径（成功/失败/panic）都会走到这里        self.inner.finish_process_start(); // 无论成功失败、panic 还是提前 return，退出作用域自动减计数——每个返回点都不用手动清理    }}struct PendingProcessStartSession { // "已注册但未确认"的会话守卫：start_process 的核心安全网    inner: Arc<Inner>, // 客户端内核：回滚时通过它操作路由表    process_id: ProcessId, // 要摘除的进程 id    state: Arc<SessionState>, // 会话状态指针：摘除前比对，防止误删同 id 的新会话    armed: bool, // armed=true 表示会话已登记进路由表、且尚未被调用方接收——drop 时必须回滚}impl Drop for PendingProcessStartSession { // 实现 Drop：守卫被丢弃 = 调用方没接住 Session    fn drop(&mut self) { // 异常退出（超时/取消/panic）都会触发这里        if self.armed { // armed=true：会话已登记进路由表且未被接收——必须回滚            self.inner.remove_session_if(&self.process_id, &self.state); // 调用方没拿到 Session 就丢弃了：自动从路由表摘除，防僵尸会话        }    }}

struct RunningProcess { // 服务端每进程状态：输出缓冲 + seq 计数器 + 生命周期标志全在这里    session: ExecCommandSession, // 底层子进程句柄（pty 或 pipe 两种形态）    tty: bool, // 是否 TTY 会话（影响信号处理和输出语义）    pipe_stdin: bool, // stdin 是否管道化——决定要不要接受 write 请求    accepted_stdin_write_ids: Arc<Mutex<AcceptedStdinWriteIds>>, // stdin 写幂等缓存：重连后客户端重试同一 write_id，不重复写入子进程    output: VecDeque<RetainedOutputChunk>, // 保留输出环形缓冲：断线重连 / resume 时用来回放    retained_bytes: usize, // 当前缓冲字节数，与 chunk 数量双限淘汰旧数据    next_seq: u64, // 单调递增序号：output / exited / closed 三类事件共用同一编号空间    exit_code: Option<i32>, // 退出码；None = 还在运行（Closed 的前置条件之一）    wake_tx: watch::Sender<u64>, // 唤醒通道：本地 process/read 轮询循环靠它感知新数据    events: ExecProcessEventLog, // 事件日志 = 有界回放 + 实时广播，本地与远程消费者各取所需    output_notify: Arc<Notify>, // 一次性通知：退出路径用它等"最后一块输出冲刷完"再发 Exited    open_streams: usize, // 仍打开的 stdout/stderr 流计数（全部关闭且已退出 → 才发 Closed）    closed: bool, // 终态标志：Closed 事件是否已发出，防止重复发送    metrics: Option<ProcessMetricGuard>, // OTEL 指标守卫：drop 时自动上报耗时与结果分类    termination_requested: bool, // 是否被主动终止——影响 success / error 的指标归类    sandbox: SandboxType, // 本进程使用的沙箱类型（None / seatbelt / bwrap …）    sandbox_denied: bool, // "疑似沙箱拦截"启发式判定结果，随 Exited 事件回传供审批/重试决策    network_proxy_handle: Option<NetworkProxyHandle>, // 每进程网络代理句柄：Closed 时关闭（第 27 讲的产物）    network_policy_shutdown: Option<CancellationToken>, // 网络策略决策回调的取消令牌：进程结束即作废未决请求}

while let Some(chunk) = receiver.recv().await { // 从子进程 stdout/stderr 的 mpsc 通道循环取块    let notification = { // 临界区开始：只改状态不发 IO，网络通告放锁外        let mut processes = inner.processes.lock().await; // 拿进程表异步锁（临界区极短：只改状态不发 IO）        let Some(entry) = processes.get_mut(&process_id) else { break }; // 进程条目已被清理：停止读取，防 use-after-remove        let ProcessEntry::Running(process) = entry else { break }; // 条目还在 Starting（启动中）：等它变成 Running 再说        let seq = process.next_seq; // 取当前序号……        process.next_seq += 1; // ……再自增：保证 output/exited/closed 全局单调递增        process.retained_bytes += chunk.len(); // 记账：当前保留缓冲总字节数        process.output.push_back(RetainedOutputChunk { seq, stream, chunk: chunk.clone() }); // 写入保留缓冲（重连回放用）        while process.retained_bytes > RETAINED_OUTPUT_BYTES_PER_PROCESS || process.output.len() > RETAINED_OUTPUT_CHUNKS_PER_PROCESS { // 双限淘汰：1MiB 或 5 万块，先到先触发            let Some(evicted) = process.output.pop_front() else { break }; // 弹出最旧一块；缓冲空了就停            process.retained_bytes = process.retained_bytes.saturating_sub(evicted.chunk.len()); // 扣减字节账，saturating 防下溢        }        let _ = process.wake_tx.send(seq); // 唤醒本地 reader（process/read 的轮询循环）        process.events.publish(ExecProcessEvent::Output(output.clone())); // 写入事件日志：回放历史 + 实时广播，双通道分发        ExecOutputDeltaNotification { process_id: process_id.clone(), seq, stream, chunk } // 组装发给远程客户端的 RPC 通告体    };    output_notify.notify_waiters(); // 通知"等待最后冲刷"的退出路径（watch_exit 里等 20ms）    if let Some(notifications) = notification_sender(&inner) {        let _ = notifications.notify(EXEC_OUTPUT_DELTA_METHOD, &notification).await; // 经共享连接的通告流发出——所有进程共用一条管道    }}

fn publish_ordered_event(&self, event: ExecProcessEvent) -> Result<bool, String> { // 远程事件入口：先按 seq 重排，再投递    let Some(seq) = event.seq() else { // 无序号事件（Failed）走快速通道        self.events.publish(event); // Failed 无序号（客户端合成）：直接发布，不参与排序        return Ok(false);    };    let mut ordered_events = self.ordered_events.lock().unwrap_or_else(std::sync::PoisonError::into_inner); // 拿会话级排序锁（poison 时取内层继续，不 panic）    if ordered_events.failure.is_some() || ordered_events.closed_published || seq <= ordered_events.last_published_seq { // 已失败 / 已关闭 / 重复或过期事件：直接丢弃，不污染状态        return Ok(false);    }    ordered_events.insert_pending(event)?; // 放入 BTreeMap 暂存区（带条数/字节双限，超限即报错）    Ok(self.publish_ready(&mut ordered_events)) // 尝试从 last_published_seq+1 开始连续排空}fn publish_ready(&self, ordered_events: &mut OrderedSessionEvents) -> bool { // 排空循环：只按严格 seq 顺序投递，绝不跳号    let mut published_closed = false; // 记录本轮是否投递了 Closed（给调用方判断终态）    loop {        let next_seq = ordered_events.last_published_seq.saturating_add(1); // 下一个期望序号        let Some(event) = ordered_events.pending.remove(&next_seq) else { break }; // 缺口未补齐：停下等待，宁可晚也不乱序        ordered_events.last_published_seq = next_seq; // 推进水位线        ordered_events.exit_published |= matches!(&event, ExecProcessEvent::Exited { .. }); // 记录 Exited 是否已投递        let is_closed = matches!(&event, ExecProcessEvent::Closed { .. }); // 判断本事件是否为终态 Closed        ordered_events.closed_published |= is_closed;        published_closed |= is_closed;        if is_closed && ordered_events.exit_published { // Exited 与 Closed 都到齐：作废该进程所有未决的网络策略决策请求            self.network_policy.cancellation.record(NetworkRequestCancellationReason::ProcessFinished); // 记录取消原因：进程结束（审计用）        }        self.events.publish(event); // 最终发布进事件日志，唤醒 unified_exec / TUI 等所有本地订阅者    } // pending 出现缺口：退出循环等待补齐    published_closed // 返回本轮是否投递了 Closed}

async fn maybe_emit_closed(process_id: ProcessId, inner: Arc<Inner>) { // 只有"已退出且所有流关闭"才发 Closed，否则静默返回    let (notification, output_notify, network_proxy_handle) = { // 临界区：凑齐终态条件后打包出锁，锁外再做 IO        let mut processes = inner.processes.lock().await; // 拿进程表异步锁（临界区极短）        let Some(ProcessEntry::Running(process)) = processes.get_mut(&process_id) else { return }; // 进程已不在表里：无事可做        if process.closed || process.open_streams != 0 || process.exit_code.is_none() { // 三条件任一不满足就返回——之后流关闭/退出时还会再调本函数            return; // 条件未齐：静默返回，等下次状态变化再试        }        process.closed = true; // 置终态标志，后续重复调用直接短路（Closed 只发一次）        if let Some(network_policy_shutdown) = process.network_policy_shutdown.take() { // 取出并置空：取消所有未决的网络策略决策回调            network_policy_shutdown.cancel(); // 进程结束：取消所有未决的网络策略决策回调（第 27 讲的闭环）        }        let seq = process.next_seq;        process.next_seq += 1; // Closed 也占一个序号——与 output/exited 同属一条排序链，客户端才能确定"它之后没有事件了"        let _ = process.wake_tx.send(seq); // 唤醒本地 reader：Closed 也要让它感知        process.events.publish(ExecProcessEvent::Closed { seq }); // 写入事件日志：本地订阅者（unified_exec）也能收到终态        (ExecClosedNotification { process_id: process_id.clone(), seq }, Arc::clone(&process.output_notify), process.network_proxy_handle.take()) // 把锁内取到的东西打包带出去，锁外再做 IO    };    if let Some(network_proxy_handle) = network_proxy_handle && let Err(err) = network_proxy_handle.shutdown().await { // 关闭每进程网络代理（在锁外！避免持锁做异步 IO）        tracing::warn!("failed to shut down executor network proxy: {err}"); // 代理关闭失败只告警不致命：进程已终态，不能因清理失败卡死    }    output_notify.notify_waiters(); // 唤醒所有等待者：输出已彻底结束    tokio::spawn(async move { // 延迟清理任务：Closed 后条目不立即删除        tokio::time::sleep(EXITED_PROCESS_RETENTION).await; // 保留 30s——窗口内迟到的 read / write 重试仍能查到会话并拿到终态，而不是"进程不存在"        let mut processes = cleanup_inner.processes.lock().await; // 拿锁做最终清理        match processes.entry(cleanup_process_id) { // 按 id 查条目：存在且已 closed 才删            Entry::Occupied(entry) => { if matches!(entry.get(), ProcessEntry::Running(process) if process.closed) { entry.remove(); } } // 只删确认 closed 的条目；若同 id 已被新会话占用则不动            Entry::Vacant(_) => {} // 已被别人删了：无事发生        }    });}

输出端到端路径（远程执行）

▼

▼

▼

▼

▼