Tmux 内部辅助

tmux 按键发送分发入口

stage 上下文: 该条目位于 tmux 内部辅助层，职责是把一次按键发送请求整理后分发给两条发送路径之一。触发时机从函数体本身不可见；就本函数可证实的范围而言，它只消费入参与若干 self 绑定方法/日志器，不直接持久化任何会话状态。作为本 stage 当前可见的首个条目，它体现的是内部发送封装而非更高层编排。

这段代码在干什么

send_keys 接收 keys、block 以及两类超时参数，先通过 _prepare_keys(...) 归一化输入并得到最终是否走阻塞路径的判定。随后它记录一次调试日志，并依据 is_blocking 在 _send_blocking_keys(...) 与 _send_non_blocking_keys(...) 之间二选一地 await 调用。本协程没有显式 return，实际产出是把发送责任委托给后续 helper，并在特定参数组合下额外输出一条说明 min_timeout_sec 会被忽略的调试日志。

接口 · 参数 / IO

(self, keys: str | list[str], block: bool = False, min_timeout_sec: float = 0.0, max_timeout_sec: float = 180.0)

• 参数: keys: str | list[str] — 待发送的按键或命令序列，传入 _prepare_keys(...) 做整理; block: bool — 请求是否阻塞等待；既参与预警日志判断，也传给 _prepare_keys(...); min_timeout_sec: float — 非阻塞路径下传给 _send_non_blocking_keys(...) 的最小等待时间；若 block 为真且大于 0，会先记录其将被忽略; max_timeout_sec: float — 阻塞路径下传给 _send_blocking_keys(...) 的最大等待时间
• 读状态: self._logger, self._prepare_keys, self._send_blocking_keys, self._send_non_blocking_keys
• 返回: 无显式返回；该协程完成时解析为 None，真正结果是完成一次 helper 委托调用与相关日志输出。
• 副作用: 当 block 为 True 且 min_timeout_sec > 0.0 时，调用 self._logger.debug(...) 记录忽略说明; 调用 self._logger.debug(...) 记录整理后的 prepared_keys 及超时参数; 按分支 await self._send_blocking_keys(...) 或 await self._send_non_blocking_keys(...)

执行流

1. 先检查 block and min_timeout_sec > 0.0；若成立，则通过 self._logger.debug(...) 记录“min_timeout_sec 会被忽略”的说明。
2. 调用 self._prepare_keys(keys=keys, block=block)，取得二元结果 prepared_keys, is_blocking，其中后者作为后续实际分支条件。
3. 无论后续是否真的使用 min_timeout_sec，都会再调用一次 self._logger.debug(...)，记录 prepared_keys、min_timeout_sec 与 max_timeout_sec。
4. 若 is_blocking 为真，则 await self._send_blocking_keys(keys=prepared_keys, max_timeout_sec=max_timeout_sec)。
5. 否则走非阻塞路径，await self._send_non_blocking_keys(keys=prepared_keys, min_timeout_sec=min_timeout_sec)。

源码

    async def send_keys(
        self,
        keys: str | list[str],
        block: bool = False,
        min_timeout_sec: float = 0.0,
        max_timeout_sec: float = 180.0,
    ):
        """
        Execute a command in the tmux session.

        Args:
            keys (str): The keys to send to the tmux session.
            block (bool): Whether to wait for the command to complete.
            min_timeout_sec (float): Minimum time in seconds to wait after executing.
                Defaults to 0.
            max_timeout_sec (float): Maximum time in seconds to wait for blocking
                commands. Defaults to 3 minutes.
        """
        if block and min_timeout_sec > 0.0:
            self._logger.debug("min_timeout_sec will be ignored because block is True.")

        prepared_keys, is_blocking = self._prepare_keys(
            keys=keys,
            block=block,
        )

        self._logger.debug(
            f"Sending keys: {prepared_keys}"
            f" min_timeout_sec: {min_timeout_sec}"
            f" max_timeout_sec: {max_timeout_sec}"
        )

        if is_blocking:
            await self._send_blocking_keys(
                keys=prepared_keys,
                max_timeout_sec=max_timeout_sec,
            )
        else:
            await self._send_non_blocking_keys(
                keys=prepared_keys,
                min_timeout_sec=min_timeout_sec,
            )

Non-obvious 设计决策

• 阻塞与否不直接只由入参 block 决定，而是以 _prepare_keys(...) 返回的 is_blocking 为准；这表明本函数把“最终发送策略判定”收口到预处理阶段，避免在当前层重复解释 keys 形式或特殊命令语义。
• 对 block=True 且 min_timeout_sec>0 的组合，代码没有报错或改写参数，只输出一条调试日志后继续执行；这是宽容处理，允许调用方传入冗余参数，同时明确提示该值不会生效。
• 第二条调试日志总是同时打印 min_timeout_sec 与 max_timeout_sec，即使稍后分支可能只使用其中一个；因此代码专门在前面加了一条“会被忽略”的日志来解释这种日志字段与实际执行路径之间的偏差。

上下游关系

• 调用方: 未在给定源码片段中展示，无法从本函数体内确认具体调用者
• 核心被调用: self._prepare_keys; self._send_blocking_keys; self._send_non_blocking_keys; self._logger.debug
• 配置/状态来源: 参数 keys; 参数 block; 参数 min_timeout_sec; 参数 max_timeout_sec
• 结果去向: prepared_keys 传入 _send_blocking_keys(...) 或 _send_non_blocking_keys(...); max_timeout_sec 仅传入 _send_blocking_keys(...); min_timeout_sec 仅传入 _send_non_blocking_keys(...); 调试信息输出到 self._logger.debug(...)

tmux 按键负载预处理助手

stage 上下文: 该函数属于 tmux 会话子系统的内部细化 helper，负责在真正发送按键前整理输入形态并计算本次返回的阻塞标志。触发点是同 stage 的 TmuxSession.send_keys，后者先调用这里得到规范化结果，再继续后续发送流程。相较已翻译的兄弟函数 send_keys，这里不执行发送，只处理 keys 与返回布尔值的预判。

这段代码在干什么

_prepare_keys 接收 keys 与 block，把字符串输入统一成列表，并依据最后一个按键是否被 _is_executing_command(keys[-1]) 识别为执行命令的按键，决定返回的阻塞标志。若 block 为假、按键列表为空，或最后一个按键不构成执行命令，则直接返回当前 keys 与 False。只有在 block 为真且最后一个按键满足执行判定时，它才先用 _prevent_execution(keys) 重写列表，再在该列表后追加 [self._TMUX_COMPLETION_COMMAND, "Enter"]，并返回该列表与 True；本函数本身不写入实例状态。

接口 · 参数 / IO

(self, keys: str | list[str], block: bool) -> tuple[list[str], bool]

• 参数: keys: str | list[str] — 待发送到终端的按键负载；若为 str 会先转换成新的单元素列表，若为 list[str] 则后续可能原样返回、被 _prevent_execution(keys) 替换，或在当前使用中的列表对象上被 extend(...) 追加元素; block: bool — 调用方请求的阻塞意图；本函数据此与末尾按键判定共同计算最终返回的阻塞标志
• 读状态: self._is_executing_command, self._prevent_execution, self._TMUX_COMPLETION_COMMAND
• 返回: 返回 tuple[list[str], bool]：第一项是整理后的按键列表，第二项是本函数计算出的阻塞标志。返回的列表可能是由字符串新建出的列表、原始列表对象，或 _prevent_execution(keys) 返回的新列表/重写结果；在进入追加分支时，函数会对当时使用的列表执行 extend([self._TMUX_COMPLETION_COMMAND, "Enter"])。
• 副作用: 不写入任何 self._ 实例字段; 在满足追加分支时，会修改当时使用中的 keys 列表对象内容（通过 extend(...)）

执行流

1. 先检查 keys 的类型；若 isinstance(keys, str) 为真，则把它包装为新的单元素列表 [keys]，把后续处理统一到列表路径。
2. 随后执行单个提前返回判定：只要 not block、not keys、或 not self._is_executing_command(keys[-1]) 三者之一成立，就直接返回当前 keys 与 False。
3. 如果没有提前返回，说明此时 block 为真、列表非空，且最后一个元素通过了 self._is_executing_command(keys[-1]) 检查；函数接着调用 self._prevent_execution(keys)，并以其返回值重新绑定 keys。
4. 在重写后的 keys 上追加 self._TMUX_COMPLETION_COMMAND 与字面量 "Enter" 两个元素，然后返回该列表与 True。

源码

    def _prepare_keys(
        self,
        keys: str | list[str],
        block: bool,
    ) -> tuple[list[str], bool]:
        """
        Prepare keys for sending to the terminal.

        Args:
            keys (str | list[str]): The keys to send to the terminal.
            block (bool): Whether to wait for the command to complete.

        Returns:
            tuple[list[str], bool]: The keys to send to the terminal and whether the
                the keys are blocking.
        """
        if isinstance(keys, str):
            keys = [keys]

        if not block or not keys or not self._is_executing_command(keys[-1]):
            return keys, False

        keys = self._prevent_execution(keys)
        keys.extend([self._TMUX_COMPLETION_COMMAND, "Enter"])

        return keys, True

Non-obvious 设计决策

• 函数采用单个复合条件的提前返回结构，把“不阻塞”的三种情况——block 为假、列表为空、末尾按键未通过执行判定——统一降级为返回布尔值 False；这一点直接体现在 if not block or not keys or not self._is_executing_command(keys[-1]): return keys, False。
• 它只在最后一个按键满足 self._is_executing_command(keys[-1]) 时才保留返回值中的阻塞标志为 True；也就是说，调用方传入 block=True 并不足以单独决定结果，代码显式要求再通过一次末尾按键检查。
• 进入阻塞分支后，代码不直接在原列表上追加，而是先执行 keys = self._prevent_execution(keys) 再追加后续元素；从源码可见，这是一次显式的重写步骤，但其具体重写内容由 helper 自身决定，这里不在本函数内展开。

上下游关系

• 调用方: TmuxSession.send_keys
• 核心被调用: self._is_executing_command; self._prevent_execution; list.extend
• 配置/状态来源: 参数 keys; 参数 block; 常量/属性 self._TMUX_COMPLETION_COMMAND
• 结果去向: 返回给 TmuxSession.send_keys 作为规范化后的 keys; 返回给 TmuxSession.send_keys 作为本函数计算出的阻塞布尔值
• 同类 sibling: TmuxSession.send_keys：先调用本函数获取 (prepared_keys, is_blocking)，自身再继续后续发送路径

判定单个按键是否触发执行

stage 上下文: 该函数属于 tmux 子系统内部的细粒度判定 helper，职责是把“这个 key 是否代表提交执行”压缩为一个布尔谓词。它位于同 stage 的按键处理辅助层级中，相比兄弟函数 send_keys 与 _prepare_keys 的编排职责，这里只负责单个 key 的最小判定，不直接操作会话或发送流程。

这段代码在干什么

_is_executing_command 接收一个 key: str，返回该字符串是否应被视为“执行命令”的输入。函数本体只做一个布尔合取判定：先测试 self._is_enter_key(key)，否则再测试 self._ends_with_newline(key)。它不写入任何实例属性，实际产出仅为布尔返回值。

接口 · 参数 / IO

(self, key: str) -> bool

• 参数: key: str — 待判定的单个 tmux 按键字符串
• 读状态: 调用 self._is_enter_key(key), 调用 self._ends_with_newline(key)
• 返回: 布尔值；当 key 被 _is_enter_key(key) 识别为 Enter 类按键，或在前者为假时被 _ends_with_newline(key) 识别为以换行结尾时，返回 True，否则返回 False。

执行流

1. 先以传入的 key 调用 self._is_enter_key(key)，检查它是否属于 Enter 风格按键。
2. 若前一步结果为真，or 表达式短路，函数立即以 True 返回，不再求值 self._ends_with_newline(key)。
3. 仅当前一步为假时，才继续调用 self._ends_with_newline(key)，检查该字符串是否以换行结束。
4. 返回这两个 helper 判定的逻辑或结果。

源码

    def _is_executing_command(self, key: str) -> bool:
        return self._is_enter_key(key) or self._ends_with_newline(key)

Non-obvious 设计决策

• 函数把“执行命令”压缩为两个条件的逻辑或：_is_enter_key(key) or _ends_with_newline(key)；这说明代码显式接受两类等价信号，而不是只认某一种输入形式。
• 这里直接使用 Python 的 or 表达式而非分别保存中间变量，带来的实际取舍是短路求值：一旦 _is_enter_key(key) 为真，就跳过 _ends_with_newline(key) 的检查。

上下游关系

• 调用方: 未在给定源码片段中出现直接调用方; 外部内部 helper 若需要把单个 key 归类为“执行命令”可调用此谓词
• 核心被调用: TmuxSession._is_enter_key; TmuxSession._ends_with_newline
• 配置/状态来源: 参数 key; self._is_enter_key 的返回结果; self._ends_with_newline 的返回结果
• 结果去向: 当前函数的布尔返回值; 调用方可据此决定是否把该 key 视为触发执行的输入
• 同类 sibling: 同 stage 兄弟 _prepare_keys 处理按键列表与阻塞策略；本函数只处理单个 key 的真值判定。; 同 stage 兄弟 send_keys 负责实际发送路径选择；本函数不参与发送，只提供一个布尔谓词。

判定按键是否属于预设成员表

stage 上下文: 该条目位于 tmux 子系统内部 helper 分层中，但就当前源码可见范围，本函数本体只承担一次简单布尔判定。它没有展示与其他 helper 的直接交互，也不包含额外控制流或状态处理。

这段代码在干什么

_is_enter_key 接收一个 key: str，读取 self._ENTER_KEYS，并直接返回 key in self._ENTER_KEYS 的布尔结果。函数唯一产出是该成员测试结果；源码中没有实例状态写入、外部调用或其他副作用。

接口 · 参数 / IO

(self, key: str) -> bool

• 参数: key: str — 待判定的按键字符串
• 读状态: self._ENTER_KEYS
• 返回: 返回 key 是否属于 self._ENTER_KEYS 的布尔值。

执行流

1. 读取实例上的 self._ENTER_KEYS。
2. 对传入的 key 执行成员测试 key in self._ENTER_KEYS。
3. 将该测试结果作为布尔值直接返回。

源码

    def _is_enter_key(self, key: str) -> bool:
        return key in self._ENTER_KEYS

Non-obvious 设计决策

• 该实现把整个判定压缩为单行成员测试，没有加入额外分支、归一化或异常处理；因此函数语义完全等同于 key in self._ENTER_KEYS 这一源码表达式。
• 函数本体只读 self._ENTER_KEYS 而不改写任何实例属性，说明它被实现为无副作用的薄包装判定接口。

上下游关系

• 调用方: 当前提供的源码片段未展示直接调用方
• 核心被调用: 无；函数体内没有调用其他函数或方法
• 配置/状态来源: self._ENTER_KEYS
• 结果去向: 直接返回给调用者的布尔结果

基于会话正则的匹配布尔封装

stage 上下文: 该条目位于 tmux 子系统内部 helper 分层中，职责很窄，只把一次正则搜索结果规整成布尔值。它属于实现细节级别的会话辅助函数，不承担状态编排，也不直接操作 tmux 外部资源。

这段代码在干什么

_ends_with_newline 接收一个 key: str，读取实例上的 self._ENDS_WITH_NEWLINE_PATTERN，并对 key 执行 re.search(...)。函数把搜索结果保存到局部变量 result，随后返回 result is not None 的布尔值。源码中没有任何实例状态写入，也没有除正则搜索外的额外处理。

接口 · 参数 / IO

(self, key: str) -> bool

• 参数: key: str — 传入 re.search(self._ENDS_WITH_NEWLINE_PATTERN, key) 的待匹配字符串
• 读状态: self._ENDS_WITH_NEWLINE_PATTERN
• 返回: 返回 re.search(self._ENDS_WITH_NEWLINE_PATTERN, key) 是否得到匹配；有匹配为 True，否则为 False。

执行流

1. 以 self._ENDS_WITH_NEWLINE_PATTERN 和参数 key 调用 re.search(...)，并将返回值存入局部变量 result。
2. 执行显式非空判断 result is not None，把匹配对象或 None 归一化为布尔返回值。

源码

    def _ends_with_newline(self, key: str) -> bool:
        result = re.search(self._ENDS_WITH_NEWLINE_PATTERN, key)
        return result is not None

Non-obvious 设计决策

• 函数没有直接返回 re.search(...) 的原始结果，而是用 result is not None 做显式布尔化；这把返回契约固定为 bool，避免向调用方暴露正则匹配对象。
• 匹配模式取自实例属性 self._ENDS_WITH_NEWLINE_PATTERN，而不是在函数体内内联字面量；可见取舍是复用对象已有配置，而非在此处重新定义规则。

上下游关系

• 调用方: 未在本翻译单元源码中直接出现；调用方信息不由此段代码显式给出
• 核心被调用: re.search
• 配置/状态来源: self._ENDS_WITH_NEWLINE_PATTERN; 参数 key
• 结果去向: 直接返回给该函数的调用方，形式为布尔值; 局部变量 result 仅作为返回前的中间承载
• 同类 sibling: 同 stage 的 _is_enter_key 也是单一判定 helper，但它做的是集合成员测试；本函数对应的是正则匹配后的布尔封装。

tmux 按键序列去执行化辅助器

stage 上下文: 该条目位于 tmux 子系统的内部 helper 层，职责是对按键 token 列表做末尾清理，使其不再满足“会触发命令执行”的尾部形态。它与同 stage 的 _is_executing_command、_is_enter_key、_ends_with_newline 形成前后配套关系：兄弟函数负责判定，当前函数负责按该判定结果重写列表。

这段代码在干什么

_prevent_execution 接收 keys: list[str]，先复制一份列表，再持续检查末尾 token 是否被 self._is_executing_command(keys[-1]) 视为执行命令的输入。若是 Enter 类 token，则直接移除；否则对最后一个 token 按 self._NEWLINE_CHARS 做右侧 rstrip，必要时回写或删除。函数返回清理后的新列表，不写入任何实例状态，也不修改调用方传入的原列表。

接口 · 参数 / IO

(self, keys: list[str]) -> list[str]

• 参数: keys: list[str] — 待清理的 tmux 按键 token 列表，函数会基于其末尾元素是否构成执行输入来重写副本
• 读状态: self._is_executing_command(...), self._is_enter_key(...), self._NEWLINE_CHARS
• 返回: 返回一个新的 list[str]：其末尾已被反复清理，直到列表为空或最后一个 token 不再被判定为执行命令。
• 副作用: 无实例状态写入; 对局部副本 keys = keys.copy() 进行修改，不改变调用方原始列表

执行流

1. 首先执行 keys = keys.copy()，建立输入列表的副本，后续所有修改都作用于这份局部列表。
2. 进入 while keys and self._is_executing_command(keys[-1]) 循环；只要列表非空且最后一个 token 仍会触发执行，就继续处理尾部。
3. 若 self._is_enter_key(keys[-1]) 为真，则把该末尾 token 直接 pop() 掉。
4. 否则进入非 Enter 分支：对最后一个 token 执行 rstrip(self._NEWLINE_CHARS)，结果保存到 stripped_key。
5. 如果 stripped_key 非空，则用它覆盖原末尾元素 keys[-1] = stripped_key；如果为空，则说明剥掉右侧换行字符后该 token 已无内容，直接 pop() 删除该元素。
6. 循环结束后返回当前 keys，此时其末尾已不再满足 _is_executing_command(...)，或列表已空。

源码

    def _prevent_execution(self, keys: list[str]) -> list[str]:
        keys = keys.copy()
        while keys and self._is_executing_command(keys[-1]):
            if self._is_enter_key(keys[-1]):
                keys.pop()
            else:
                stripped_key = keys[-1].rstrip(self._NEWLINE_CHARS)

                if stripped_key:
                    keys[-1] = stripped_key
                else:
                    keys.pop()

        return keys

Non-obvious 设计决策

• keys.copy() 明确选择在副本上工作，而不是原地修改传入参数；这一点从首行赋值可直接观察到，避免了调用方列表被隐式改写。
• 使用 while 而非单次 if，意味着代码显式支持连续清理多个尾部执行触发项，而不是只处理最后一个 token 一次。
• Enter 类 token 与非 Enter token 走分支处理：前者整项删除，后者仅按 self._NEWLINE_CHARS 做右侧剥离；这保持了非 Enter token 在去掉尾部换行后的其余内容。
• 对 rstrip(...) 结果做显式真值检查，说明代码区分“剥离后仍有内容”和“剥离后为空串”两种情况；为空时改为删除整个 token，避免把空字符串留在列表尾部。

上下游关系

• 调用方: 未在给定源码片段中展示; 外部调用方未知；本片段只能确认它是实例内部方法
• 核心被调用: self._is_executing_command(keys[-1]); self._is_enter_key(keys[-1]); str.rstrip(self._NEWLINE_CHARS); list.copy(); list.pop()
• 配置/状态来源: self._NEWLINE_CHARS; 由 self._is_executing_command(...) 封装的对象内判定逻辑; 由 self._is_enter_key(...) 封装的对象内判定逻辑
• 结果去向: 返回给本方法的直接调用者作为清理后的按键列表; 作为同一 TmuxSession 内部后续处理可消费的列表值
• 同类 sibling: TmuxSession._is_executing_command：负责判定单个 token 是否构成执行输入；本函数据此决定是否继续清理尾部。; TmuxSession._is_enter_key：负责识别 Enter 类 token；本函数在 Enter 分支中直接删除该 token。; TmuxSession._ends_with_newline：同属执行判定链上的基础检查，但本函数自身不直接调用它，而是经由 _is_executing_command(...) 间接利用该语义。

tmux 非阻塞按键发送与最小时延补偿

stage 上下文: 该条目属于 tmux 子系统内部 helper，用于把一组已准备好的命令 token 展开成底层命令并逐条提交给 environment.exec(...)。它只覆盖非阻塞分支内部的实际执行与节流补偿，不处理按键预处理、阻塞判定或完成检测；这些职责由同 stage 的其他兄弟函数承担。

这段代码在干什么

_send_non_blocking_keys(...) 接收 keys 与 min_timeout_sec，先调用 self._tmux_send_keys(keys) 生成命令序列，再按顺序以 user=self._user 逐条执行 self.environment.exec(...)。每次执行后都会检查结果对象的 return_code；一旦非零，就用同一结果里的 stderr 组装 RuntimeError 抛出。若前面这段执行累计耗时仍小于 min_timeout_sec，函数会额外 await asyncio.sleep(...) 补足剩余时长；否则直接结束，本身无显式返回值，也不写入实例状态。

接口 · 参数 / IO

(self, keys: list[str], min_timeout_sec: float) -> None

• 参数: keys: list[str] — 传入 self._tmux_send_keys(keys) 以生成待执行命令序列的按键 token 列表; min_timeout_sec: float — 与执行前后测得的耗时比较，决定是否需要额外睡眠补足最小时长
• 读状态: self._tmux_send_keys, self.environment, self._user
• 返回: 无显式返回值；真正产出是逐条执行 self._tmux_send_keys(keys) 生成的命令，并在需要时追加一次异步睡眠。
• 副作用: 通过 self.environment.exec(command=command, user=self._user) 逐条执行生成出的命令; 读取每次执行结果对象的 return_code 与 stderr 以决定是否抛出异常; 在 elapsed_time_sec < min_timeout_sec 时执行一次 asyncio.sleep(...)

执行流

1. 函数开始时记录 start_time_sec = time.time()，把它作为后续最小时长判断的起点。
2. 对 self._tmux_send_keys(keys) 产出的每个 command 逐条循环，调用 await self.environment.exec(command=command, user=self._user) 执行，并接收结果对象 result。
3. 每次拿到 result 后立即检查 result.return_code != 0；若条件成立，则以 self.environment.session_id 和 result.stderr 组装错误消息并抛出 RuntimeError，函数在该处提前终止。
4. 若所有命令都未触发上述异常，则计算 elapsed_time_sec = time.time() - start_time_sec，这里测得的是从函数开头到全部命令执行结束这一段已消耗的时间。
5. 仅当 elapsed_time_sec < min_timeout_sec 时，才 await asyncio.sleep(min_timeout_sec - elapsed_time_sec) 补足剩余时长；若耗时等于或超过阈值，由于条件是严格的小于号，睡眠会被完全跳过。

源码

    async def _send_non_blocking_keys(
        self,
        keys: list[str],
        min_timeout_sec: float,
    ):
        start_time_sec = time.time()

        for command in self._tmux_send_keys(keys):
            result = await self.environment.exec(command=command, user=self._user)
            if result.return_code != 0:
                raise RuntimeError(
                    f"{self.environment.session_id}: failed to send non-blocking keys: {result.stderr}"
                )

        elapsed_time_sec = time.time() - start_time_sec

        if elapsed_time_sec < min_timeout_sec:
            await asyncio.sleep(min_timeout_sec - elapsed_time_sec)

Non-obvious 设计决策

• 错误处理采用逐条执行后立即检查 result.return_code 的失败即抛出策略，而不是收集全部结果后统一处理；这样一旦某条命令失败，后续命令不会继续提交，且异常消息直接复用当前 result.stderr。
• 最小时长控制是“补差额”而不是固定追加等待：代码先测量从 start_time_sec 到命令循环结束的已耗时，再只在不足 min_timeout_sec 时睡眠剩余部分，因此不会无条件增加一个完整的后置延迟。
• 睡眠条件使用严格的 elapsed_time_sec < min_timeout_sec，这意味着当已耗时恰好达到阈值时也不会进入睡眠，体现的是下限满足即结束，而非追求额外缓冲。

上下游关系

• 调用方: 同一 tmux 子系统中的更高层非阻塞发送路径会 await 本协程以执行生成命令
• 核心被调用: self._tmux_send_keys; self.environment.exec; time.time; asyncio.sleep
• 配置/状态来源: 参数 keys; 参数 min_timeout_sec; self._user; self.environment.session_id
• 结果去向: 通过 self.environment.exec(...) 向外部环境提交命令; 失败时抛出 RuntimeError 给上层调用者; 成功路径在必要时通过 asyncio.sleep(...) 延后协程完成时刻
• 同类 sibling: 同 stage 的 send_keys 负责更高层发送入口；本函数只实现其非阻塞执行细节。; 同 stage 的 _prepare_keys 负责按键列表归一化与阻塞相关改写；本函数不处理这些输入整形。; 同 stage 的 _prevent_execution、_is_executing_command 等兄弟函数负责按键语义判断；本函数只消费最终传入的 keys。

tmux 阻塞式按键发送与等待助手

stage 上下文: 该条目位于 tmux internal helpers，职责是执行一次“发送按键后等待完成信号”的底层阻塞流程。它与同 stage 的 _send_non_blocking_keys 形成对照：两者都负责下发 tmux 按键命令，但这里只有在发送完成后额外执行一次 tmux wait done 等待，并记录阻塞耗时日志。

这段代码在干什么

_send_blocking_keys 接收 keys 与 max_timeout_sec，先把 keys 交给 self._tmux_send_keys(keys) 展开为一个或多个实际命令，并逐条通过 self.environment.exec(..., user=self._user) 发送。任一发送命令返回非零 return_code 时，函数立即抛出带 self.environment.session_id 和 stderr 的 RuntimeError。发送全部成功后，它执行 timeout {max_timeout_sec}s tmux wait done；对这一步的任何非零返回统一映射为 TimeoutError。若等待步骤返回零，则计算从入口开始的总耗时并写入一条 debug 日志，本身无显式返回值。

接口 · 参数 / IO

(self, keys: list[str], max_timeout_sec: float)

• 参数: keys: list[str] — 待发送到 tmux 的按键 token 列表; max_timeout_sec: float — 等待 tmux wait done 的超时上限，直接插入 timeout 命令字符串
• 读状态: self._tmux_send_keys, self.environment, self.environment.session_id, self._user, self._logger
• 返回: 无显式返回值；真正产出是向外部环境发送按键命令、执行一次等待命令，并在失败时抛出异常。
• 副作用: 调用 self.environment.exec(...) 执行展开后的 tmux 发送命令; 调用 self.environment.exec(...) 执行 timeout {max_timeout_sec}s tmux wait done; 发送失败时抛出 RuntimeError; 等待步骤非零返回时抛出 TimeoutError; 成功路径写入一条 debug 日志

执行流

1. 记录入口时间 start_time_sec = time.time()，用于整个阻塞流程结束后的耗时统计。
2. 调用 self._tmux_send_keys(keys) 取得待执行命令序列，并对其中每个 command 逐条 await self.environment.exec(command=command, user=self._user)。
3. 每次发送后检查 result.return_code；只要某条发送命令返回非零，就立即抛出 RuntimeError(f"{self.environment.session_id}: failed to send blocking keys: {result.stderr}")，后续命令与等待步骤都不会继续执行。
4. 若全部发送命令成功，再执行 await self.environment.exec(f"timeout {max_timeout_sec}s tmux wait done", user=self._user) 进入阻塞等待。
5. 检查等待命令的 result.return_code；代码把任何非零返回统一视为超时路径，并抛出 TimeoutError(f"Command timed out after {max_timeout_sec} seconds")，不再区分其他失败原因。
6. 仅当等待命令返回零时，计算 time.time() - start_time_sec 得到总耗时，并通过 self._logger.debug(...) 记录 Blocking command completed in ...s.。

源码

    async def _send_blocking_keys(
        self,
        keys: list[str],
        max_timeout_sec: float,
    ):
        start_time_sec = time.time()

        for command in self._tmux_send_keys(keys):
            result = await self.environment.exec(command=command, user=self._user)
            if result.return_code != 0:
                raise RuntimeError(
                    f"{self.environment.session_id}: failed to send blocking keys: {result.stderr}"
                )

        result = await self.environment.exec(
            f"timeout {max_timeout_sec}s tmux wait done", user=self._user
        )
        if result.return_code != 0:
            raise TimeoutError(f"Command timed out after {max_timeout_sec} seconds")

        elapsed_time_sec = time.time() - start_time_sec
        self._logger.debug(f"Blocking command completed in {elapsed_time_sec:.2f}s.")

Non-obvious 设计决策

• 发送阶段采用逐条执行、逐条校验的失败即停策略：循环内一旦发现 return_code != 0 就直接抛出 RuntimeError。这保证调用方不会在已知发送失败后继续进入等待步骤，但代价是不会尝试继续发送剩余命令。
• 等待阶段不解析 stderr、也不细分退出原因，而是把 timeout {max_timeout_sec}s tmux wait done 的任何非零返回统一映射为 TimeoutError。这是代码里明确可见的归一化处理，换来的结果是异常类型简单，但会丢失“非零是否真由超时引起”的区分信息。
• 耗时统计覆盖的是从首次发送前到等待完成后的整段流程，因为 start_time_sec 在任何外部命令执行之前记录，最终日志反映的是“发送 + 等待”的总耗时，而非仅等待耗时。

上下游关系

• 调用方: 未知；给定源码片段未展示直接调用点; 任何需要以阻塞方式发送 tmux 按键的同类内部流程
• 核心被调用: self._tmux_send_keys; self.environment.exec; time.time; self._logger.debug
• 配置/状态来源: 参数 keys; 参数 max_timeout_sec; self._user; self.environment.session_id
• 结果去向: 外部 tmux/环境命令执行副作用; RuntimeError 异常路径; TimeoutError 异常路径; debug 日志输出
• 同类 sibling: 与 _send_non_blocking_keys 同样负责把 keys 经 _tmux_send_keys(...) 展开后逐条执行，并在发送命令非零返回时抛出异常；区别是本函数随后还会执行一次 timeout ... tmux wait done，并在成功后记录阻塞耗时。

tmux 按键命令分批构造器

stage 上下文: 该函数位于 tmux 子系统内部 helper 层，负责把上游已经准备好的按键 token 序列翻译成可执行的 tmux send-keys 命令字符串。它通常由 _send_non_blocking_keys(...) 与 _send_blocking_keys(...) 在真正调用 environment.exec(...) 前触发。与兄弟函数相比，_prepare_keys(...) 决定“发什么、是否阻塞”，而本函数只负责“怎样把这些按键打包成 tmux 能接受的命令”。

这段代码在干什么

_tmux_send_keys 接收 keys: list[str]，读取 self._session_name 与 self._TMUX_SEND_KEYS_MAX_COMMAND_LENGTH，生成一个或多个 tmux send-keys -t <session> ... 命令字符串并返回。若所有按键在 shell 转义后仍能落入单条命令长度上限，则直接返回单元素列表；否则按长度上限分批，并在单个按键本身过长时借助 _split_key_for_tmux(...) 再拆分。函数不执行命令、不写入实例状态，真正的产出是供后续发送路径逐条执行的字符串列表。

接口 · 参数 / IO

(self, keys: list[str]) -> list[str]

• 参数: keys: list[str] — 待发送到 tmux 会话的一组按键 token，由上游发送路径或阻塞包装逻辑提供
• 读状态: self._session_name, self._TMUX_SEND_KEYS_MAX_COMMAND_LENGTH, self._split_key_for_tmux
• 返回: 返回 list[str]：一个或多个可直接执行的 tmux send-keys -t <session> ... 命令字符串；顺序保持输入按键的发送顺序。
• 副作用: 无实例状态写入; 无外部命令执行; 仅构造并返回命令字符串

执行流

1. 先以 self._session_name 经过 shlex.quote(...) 后构造固定前缀 tmux send-keys -t ...，并读取长度上限 self._TMUX_SEND_KEYS_MAX_COMMAND_LENGTH 作为后续所有打包判定的基准。
2. 把全部 keys 先映射成 escaped_keys = [shlex.quote(key) for key in keys]，拼出单条候选命令 single；若其总长度未超过 max_len，立即返回 [single]，走单命令快速路径。
3. 若单条命令超限，则初始化 commands、当前批次 current_escaped 与累计长度 current_len，并定义局部 _flush()：只要当前批次非空，就把它封装成一条完整命令追加到 commands，随后清空批次并把长度重置为前缀长度。
4. 按原顺序遍历每个 key，对其重新做一次 shlex.quote(...)，并计算把该键加入当前命令所需增加的长度 addition = 1 + len(escaped)；其中额外的 1 对应前置空格。
5. 若当前批次还能容纳该键，则直接把转义后的键追加到 current_escaped；若当前批次放不下、但该键单独作为一条命令仍未超限，则先 _flush() 结束上一批，再以该键开启新批次。
6. 若某个键连单独成命令都超限，则先 _flush()，再计算可供单个转义片段使用的最大长度 max_escaped = max_len - len(prefix) - 1，并调用 self._split_key_for_tmux(key, max_escaped) 把该键拆成多个已适配长度约束的转义片段。
7. 对拆出的每个 chunk_escaped，继续复用同一套“能装入当前批次就追加，否则先 _flush() 再新开一条”的策略，把超长单键分摊到一条或多条 tmux send-keys 命令中。
8. 遍历结束后执行最后一次 _flush()，把尾部尚未落盘的批次写入 commands，再返回完整命令列表。

源码

    def _tmux_send_keys(self, keys: list[str]) -> list[str]:
        """Build one or more ``tmux send-keys`` commands for *keys*.

        If the shell-escaped command would exceed the tmux command-length
        limit, the keys are spread across multiple commands so that each
        individual command stays within the limit.  Oversized single keys
        are split into sub-strings whose quoted form fits.
        """
        prefix = "tmux send-keys -t " + shlex.quote(self._session_name)
        max_len = self._TMUX_SEND_KEYS_MAX_COMMAND_LENGTH

        escaped_keys = [shlex.quote(key) for key in keys]
        single = prefix + " " + " ".join(escaped_keys)
        if len(single) <= max_len:
            return [single]

        commands: list[str] = []
        current_escaped: list[str] = []
        current_len = len(prefix)

        def _flush() -> None:
            nonlocal current_len
            if current_escaped:
                commands.append(prefix + " " + " ".join(current_escaped))
                current_escaped.clear()
                current_len = len(prefix)

        for key in keys:
            escaped = shlex.quote(key)
            addition = 1 + len(escaped)  # space + quoted key

            if current_len + addition <= max_len:
                current_escaped.append(escaped)
                current_len += addition
            elif len(prefix) + addition <= max_len:
                _flush()
                current_escaped.append(escaped)
                current_len = len(prefix) + addition
            else:
                _flush()
                max_escaped = max_len - len(prefix) - 1
                for chunk_escaped in self._split_key_for_tmux(key, max_escaped):
                    if current_len + 1 + len(chunk_escaped) <= max_len:
                        current_escaped.append(chunk_escaped)
                        current_len += 1 + len(chunk_escaped)
                    else:
                        _flush()
                        current_escaped.append(chunk_escaped)
                        current_len = len(prefix) + 1 + len(chunk_escaped)

        _flush()
        return commands

Non-obvious 设计决策

• 函数先尝试一次“整体单命令”快速路径，而不是一开始就进入逐键分批流程；这是对常见短命令场景的优化，既减少分支，也避免在无需拆分时维护批次状态。
• 长度判定基于 shlex.quote(...) 后的字符串，而不是原始 key 长度；代码显式围绕 shell 转义后的真实命令长度做预算，避免原始文本较短但转义后膨胀，最终生成超出 tmux 限制的命令。
• 对超限情况采用“两级退让”：先尝试按键级分批，只有当单个键自身仍过长时才调用 _split_key_for_tmux(...)。这种分层策略保留了普通按键的原子性，避免把本可整体发送的 token 过度拆碎。
• 局部 _flush() 只在 current_escaped 非空时追加命令，意味着即便遇到连续边界切换或超长键分裂，也不会生成空的 tmux send-keys 命令；这是一个静默但重要的正确性约束。

上下游关系

• 调用方: TmuxSession._send_non_blocking_keys; TmuxSession._send_blocking_keys; tmux 内部按键发送路径在实际 environment.exec(...) 前的命令准备步骤; 由 TmuxSession.send_keys 间接触发的阻塞/非阻塞发送分支
• 核心被调用: shlex.quote; TmuxSession._split_key_for_tmux
• 配置/状态来源: self._session_name：决定 -t <session> 目标会话; self._TMUX_SEND_KEYS_MAX_COMMAND_LENGTH：决定单条命令最大允许长度; 输入参数 keys：提供待编码的按键序列; tmux send-keys 命令格式约束：决定返回值字符串的目标形态
• 结果去向: 返回给 TmuxSession._send_non_blocking_keys 逐条执行; 返回给 TmuxSession._send_blocking_keys 逐条执行; 其返回列表中的每个字符串随后作为 environment.exec(...) 的命令参数; 为上层阻塞/非阻塞按键发送流程提供已分批且不超长的 tmux 命令表示
• 同类 sibling: 相较于 _prepare_keys(...) 负责阻塞判定与补尾命令，本函数只处理命令字符串的长度约束与分批。; 相较于 _send_non_blocking_keys(...) 与 _send_blocking_keys(...) 会真正调用 environment.exec(...) 并处理错误，本函数纯构造返回值，没有执行副作用。; 当 _prepare_keys(...) 已经追加 self._TMUX_COMPLETION_COMMAND 与 Enter 后，这些新增 token 也会被本函数纳入同一套分批规则。

tmux 发送命令分批刷出助手

stage 上下文: 该单元是 TmuxSession._tmux_send_keys 内部的嵌套助手，只服务于 tmux send-keys 命令字符串的分批拼装。它在外围逻辑判断“当前批次需要落盘”或“循环结束后做最终排空”时被触发，把已缓冲的一批按键转成一条完整命令。与同 stage 的 _tmux_send_keys、_split_key_for_tmux 一样，它属于底层实现细节；不同于 _send_non_blocking_keys / _send_blocking_keys，它不执行任何命令。

这段代码在干什么

_flush 从闭包中读取 current_escaped、commands、prefix 与 current_len，负责把当前已缓冲的转义按键 token 一次性追加到累计命令列表 commands。只有在 current_escaped 非空时，它才生成 prefix + " " + " ".join(current_escaped) 这一条 tmux 命令，并随后清空缓冲、把 current_len 重置为 len(prefix)。函数返回 None；真正产出是对外围局部状态的原地更新，供 _tmux_send_keys 继续分批或最终返回。

接口 · 参数 / IO

() -> None

• 返回: 返回 None；真正结果是把一条组装完成的命令写入闭包变量 commands，并重置当前批次缓冲状态。
• 副作用: 向闭包列表 commands 追加一条命令字符串; 清空闭包列表 current_escaped; 通过 nonlocal current_len 回写当前长度为 len(prefix)

执行流

1. 声明 nonlocal current_len，表明后续会直接改写外围 _tmux_send_keys 中维护的当前命令长度计数。
2. 检查闭包变量 current_escaped 是否非空；若当前没有任何待输出 token，则函数直接结束，不产生新命令。
3. 当 current_escaped 有内容时，将 prefix 与 ' '.join(current_escaped) 组合成一条完整 shell 命令，并追加到 commands。
4. 完成追加后，调用 current_escaped.clear() 清空当前批次缓冲，避免这些 token 被重复写入下一条命令。
5. 最后把 current_len 重置为 len(prefix)，使外围逻辑后续按“只有前缀、尚未追加任何 key”的长度基线继续累计。

源码

        def _flush() -> None:
            nonlocal current_len
            if current_escaped:
                commands.append(prefix + " " + " ".join(current_escaped))
                current_escaped.clear()
                current_len = len(prefix)

Non-obvious 设计决策

• 它只在 current_escaped 非空时才追加命令，这是一个显式的空批次抑制策略：避免生成仅含 prefix 的空 tmux send-keys 命令，减少无意义发送。
• 函数在刷出后同时执行“清空缓冲”和“重置 current_len”，把一次批次提交后的两项收尾动作绑定在同一原子助手里；这样外围 _tmux_send_keys 在多处触发排空时无需重复维护这两项一致性。
• 这里不返回新列表或新长度，而是通过闭包原地改写 commands、current_escaped 与 current_len。这种选择符合其作为微型 batching primitive 的定位，降低了外围分批循环的返回值拆包复杂度。

上下游关系

• 调用方: TmuxSession._tmux_send_keys 中的分批构造循环; TmuxSession._tmux_send_keys 的最终排空路径
• 核心被调用: commands.append(...); ' '.join(current_escaped); current_escaped.clear(); len(prefix)
• 配置/状态来源: 闭包变量 prefix：当前 tmux send-keys 命令前缀; 闭包变量 current_escaped：当前批次已转义 token 缓冲; 闭包变量 commands：累计输出命令列表; 闭包变量 current_len：外围维护的当前命令长度计数
• 结果去向: 闭包列表 commands，供 TmuxSession._tmux_send_keys 最终返回; 闭包变量 current_escaped 的空状态，供外围开始下一批收集; 闭包变量 current_len 的重置值，供外围继续长度控制; 后续 _send_non_blocking_keys / _send_blocking_keys 消费的 tmux 命令序列
• 同类 sibling: 与兄弟函数 _tmux_send_keys 相比，_flush 不负责拆分策略，只负责把当前已决定的一批 token 落成一条命令。; 与 _prepare_keys、_prevent_execution 这类上游按键规范化函数不同，本函数完全不解释按键语义，只处理命令批次缓冲的提交。

tmux 按键分块转义辅助

stage 上下文: 该函数位于 tmux 子系统内部辅助层，职责是为下游 tmux send-keys 命令构造提供长度受限的转义片段。它不会直接发送命令，只在调用方发现单条按键经 shell 转义后可能超长时触发。相较同 stage 的 _tmux_send_keys 负责整批命令拼装、_send_*_keys 负责真正执行，本函数只处理“单个过长 key 如何切块”这一更底层问题。

这段代码在干什么

_split_key_for_tmux 接收原始字符串 key 与转义后长度上限 max_escaped_len，把 key 拆成多个已经过 shlex.quote(...) 处理的片段，并保证每个片段的转义结果长度都不超过给定上限。函数只读取这两个入参，不访问任何 self 状态，也不产生外部副作用。返回值是 list[str]，供 _tmux_send_keys 继续拼接为一条或多条 tmux send-keys 命令。

接口 · 参数 / IO

(key: str, max_escaped_len: int) -> list[str]

• 参数: key: str — 待拆分的单个按键/文本串；来自上游 _tmux_send_keys 处理中判定为过长的输入; max_escaped_len: int — 单个片段在执行 shlex.quote(...) 后允许达到的最大长度
• 返回: 返回按顺序组成的 list[str]；其中每个元素都是对原字符串某一连续子串执行 shlex.quote(...) 后的结果，供调用方继续组装 tmux 命令。

执行流

1. 初始化空列表 chunks 作为累计输出，并把完整输入放入局部变量 remaining，后续循环始终围绕“尚未切出的尾部文本”推进。
2. 只要 remaining 非空，就为当前这段文本建立一次前缀搜索：把候选长度边界设为 lo = 1、hi = len(remaining)，并用 best = 1 记录当前已知可接受的最大前缀长度。
3. 在内部循环中反复取中点 mid，对 remaining[:mid] 执行 shlex.quote(...) 并检查其长度是否 <= max_escaped_len；若满足上限，就把 best 更新为 mid 并继续向更长前缀搜索，否则缩小到更短前缀区间。
4. 当本轮二分结束后，把 remaining[:best] 的转义结果追加到 chunks，这一步产出当前能够放入单个 tmux 参数中的最大合法片段。
5. 随后将 remaining 截去已消费的前 best 个字符，即更新为 remaining[best:]，进入下一轮直到原始 key 被完全覆盖。
6. 循环结束后返回 chunks；返回列表中的元素顺序与原始文本顺序一致，函数本身不写入实例字段，也不执行任何 tmux 或环境命令。

源码

    def _split_key_for_tmux(key: str, max_escaped_len: int) -> list[str]:
        """Split *key* into ``shlex.quote``-d chunks each ≤ *max_escaped_len*."""
        chunks: list[str] = []
        remaining = key
        while remaining:
            lo, hi, best = 1, len(remaining), 1
            while lo <= hi:
                mid = (lo + hi) // 2
                if len(shlex.quote(remaining[:mid])) <= max_escaped_len:
                    best = mid
                    lo = mid + 1
                else:
                    hi = mid - 1
            chunks.append(shlex.quote(remaining[:best]))
            remaining = remaining[best:]
        return chunks

Non-obvious 设计决策

• 长度约束是以 len(shlex.quote(...)) 为准，而不是原始子串长度。这一取舍直接体现在内部判定条件 len(shlex.quote(remaining[:mid]))) <= max_escaped_len，目的是把 shell 转义引入的额外字符也计入限制；否则原始文本看似够短，但实际拼进命令后仍可能超长。
• 每轮选择“当前剩余文本中可接受的最长前缀”，并通过二分搜索而非线性试探得到 best。代码层面体现为 lo/hi/mid 的区间收缩；这样做减少了对 shlex.quote(...) 的重复测量次数，适合被 _tmux_send_keys 在命令长度控制路径中频繁调用。
• 函数返回的是已经转义好的片段，而不是原始片段。这与其注释 Split *key* into shlex.quote-d chunks 以及两次 chunks.append(shlex.quote(...)) / 长度测量保持一致，可避免调用方再次逐片转义时出现长度重新变化，保证上限判断与实际输出一致。

上下游关系

• 调用方: TmuxSession._tmux_send_keys：当单个 key 经转义后无法直接装入当前 tmux send-keys 命令长度预算时调用本函数拆分
• 核心被调用: shlex.quote：测量候选前缀的转义后长度; shlex.quote：生成最终写入 chunks 的 shell 安全片段; len：计算 remaining 与转义字符串的长度; list.append：把每轮确定的片段追加到输出列表
• 配置/状态来源: 参数 key：待分块的原始文本来源; 参数 max_escaped_len：调用方施加的长度上限; 局部变量 remaining：当前尚未拆分的尾部内容; 局部变量 best：本轮可接受前缀长度的暂存结果
• 结果去向: 返回给 TmuxSession._tmux_send_keys，作为其分批构造 tmux send-keys -t <session> ... 命令的一部分; 间接服务于 TmuxSession._send_non_blocking_keys，后者会执行 _tmux_send_keys 产出的命令列表; 间接服务于 TmuxSession._send_blocking_keys，后者同样依赖 _tmux_send_keys 的命令展开结果; 与同 stage 的 _tmux_send_keys._flush 配合：本函数先把超长单键拆成合法片段，_flush 再把这些片段打包进最终命令字符串
• 同类 sibling: 相对 _tmux_send_keys：后者负责整条 tmux 命令的分批与前缀拼装，本函数只解决单个 key 过长时的细粒度拆分。; 相对 _send_non_blocking_keys / _send_blocking_keys：这些兄弟函数会真正执行环境命令并处理失败/超时，本函数纯计算、无 I/O、副作用为空。

tmux 依赖探测与补装助手

stage 上下文: 该条目属于 tmux 子系统内部辅助逻辑，负责在实际会话建立前补齐远端所需二进制。它不直接发送 tmux 按键，也不操作会话内状态，而是围绕环境探测、安装分支选择与安装后校验展开。与同 stage 中偏命令构造/发送的兄弟函数不同，这里唯一处理的是工具可用性保障。

这段代码在干什么

_attempt_tmux_installation 先在远端检查 tmux 是否已存在，并仅在 self._remote_asciinema_recording_path is not None 时额外检查 asciinema。若所需工具都已可用，则记录 debug 日志后直接返回；否则根据 _detect_system_info() 返回的 system_info['package_manager'] 尝试构造一次性联合安装命令，并在成功后分别复验每个仍缺失的工具。若联合安装不可走、执行失败，或执行成功但单项复验失败，则分别退回到源码编译 tmux 与 pip 安装 asciinema；函数返回 None，实际产出是远端命令执行、日志输出与可能发生的安装副作用。

接口 · 参数 / IO

(self) -> None

• 参数: self: ? — 提供远端执行入口 environment.exec(...)、录屏配置 self._remote_asciinema_recording_path、日志器，以及系统探测/安装辅助方法
• 读状态: self.environment, self._remote_asciinema_recording_path, self._logger
• 返回: 返回 None；无直接 self 状态或 register 写入，真正产出是对远端环境执行检查/安装命令、调用后备安装辅助方法，以及记录日志。
• 副作用: 以 user="root" 执行远端命令 tmux -V、asciinema --version 与可能的安装命令; 调用 self._detect_system_info() 获取系统信息并读取其中的 system_info['package_manager'] 参与分支选择; 可能调用 self._get_combined_install_command(system_info, tools_needed) 生成联合安装命令; 可能调用 self._build_tmux_from_source() 在远端编译安装 tmux; 可能调用 self._install_asciinema_with_pip() 在远端通过 pip 安装 asciinema; 写出 debug / warning 日志; 不直接写入任何 self._... 成员，也不直接读写题述 state_registers

执行流

1. 先通过 self.environment.exec(command="tmux -V", user="root") 探测 tmux，依据 return_code == 0 计算 tmux_installed。随后以 self._remote_asciinema_recording_path is not None 计算 needs_asciinema：若为真，再执行 asciinema --version 检查；若为假，则直接把 asciinema_installed = True，使后续“所需工具都齐备”判断可统一成立。
2. 若 tmux_installed and asciinema_installed 为真，则记录 Both tmux and asciinema are already installed 的 debug 日志并立即返回，不进入任何安装路径。
3. 否则根据缺失情况构造 tools_needed 列表：缺 tmux 时加入 "tmux"，缺且需要 asciinema 时加入 "asciinema"，再记录 Installing: ... 的 debug 日志。
4. 调用 self._detect_system_info() 获取 system_info；只有当 system_info["package_manager"] 为真时，才进一步调用 self._get_combined_install_command(system_info, tools_needed) 生成联合安装命令。若该命令存在，则记录所用包管理器与命令文本，并以 root 执行该安装命令。
5. 当联合安装命令执行结果 result.return_code == 0 时，函数不会立刻信任安装结果，而是仅对先前缺失的工具做复验：tmux 通过再次执行 tmux -V，asciinema 通过再次执行 asciinema --version。任一复验失败时分别记录 warning，并触发对应的单项后备安装：tmux 走 _build_tmux_from_source()，asciinema 走 _install_asciinema_with_pip()。这一轮处理完成后直接 return。
6. 如果没有可用的包管理器、未拿到联合安装命令，或联合安装命令执行返回非零，则不会在该分支内抛错或复验，而是落入末尾的逐项后备路径：缺 tmux 时记录 Installing tmux from source... 并调用 _build_tmux_from_source()；需要且缺 asciinema 时记录 Installing asciinema via pip... 并调用 _install_asciinema_with_pip()。

源码

    async def _attempt_tmux_installation(self) -> None:
        """
        Install both tmux and asciinema in a single operation for efficiency.
        """
        # Check what's already installed
        tmux_result = await self.environment.exec(command="tmux -V", user="root")
        tmux_installed = tmux_result.return_code == 0

        needs_asciinema = self._remote_asciinema_recording_path is not None
        if needs_asciinema:
            asciinema_result = await self.environment.exec(
                command="asciinema --version", user="root"
            )
            asciinema_installed = asciinema_result.return_code == 0
        else:
            asciinema_installed = True

        if tmux_installed and asciinema_installed:
            self._logger.debug("Both tmux and asciinema are already installed")
            return

        tools_needed = []
        if not tmux_installed:
            tools_needed.append("tmux")
        if needs_asciinema and not asciinema_installed:
            tools_needed.append("asciinema")

        self._logger.debug(f"Installing: {', '.join(tools_needed)}")

        # Detect system and package manager
        system_info = await self._detect_system_info()

        if system_info["package_manager"]:
            install_command = self._get_combined_install_command(
                system_info, tools_needed
            )
            if install_command:
                self._logger.debug(
                    f"Installing tools using {system_info['package_manager']}: {install_command}"
                )
                result = await self.environment.exec(
                    command=install_command, user="root"
                )

                if result.return_code == 0:
                    # Verify installations
                    if not tmux_installed:
                        verify_tmux = await self.environment.exec(
                            command="tmux -V", user="root"
                        )
                        if verify_tmux.return_code != 0:
                            self._logger.warning(
                                "tmux installation verification failed"
                            )
                            await self._build_tmux_from_source()

                    if needs_asciinema and not asciinema_installed:
                        verify_asciinema = await self.environment.exec(
                            command="asciinema --version", user="root"
                        )
                        if verify_asciinema.return_code != 0:
                            self._logger.warning(
                                "asciinema installation verification failed"
                            )
                            await self._install_asciinema_with_pip()

                    return

        # Fallback to individual installations
        if not tmux_installed:
            self._logger.warning("Installing tmux from source...")
            await self._build_tmux_from_source()

        if needs_asciinema and not asciinema_installed:
            self._logger.warning("Installing asciinema via pip...")
            await self._install_asciinema_with_pip()

Non-obvious 设计决策

• 函数把 needs_asciinema == False 显式映射为 asciinema_installed = True，从而让早退条件始终统一写成 tmux_installed and asciinema_installed，而不是在多个位置重复判断“是否需要 asciinema”。这是一种控制流归一化，代码中通过 else: asciinema_installed = True 直接体现。
• 当 system_info['package_manager'] 可用时，代码优先尝试 _get_combined_install_command(system_info, tools_needed) 所代表的一次性联合安装；其后仍对每个原本缺失的工具单独复验。这个取舍表明：包管理器安装被当作首选路径，但函数并不把命令返回零视为充分条件，而是保留逐工具兜底。
• 联合安装分支只有在 result.return_code == 0 时才进入复验并 return；若命令不可构造、包管理器不存在，或安装命令本身失败，函数直接落到末尾的单项后备安装，而不是在中间抛异常。这说明这里的策略是“尽量补装成功”，而不是把包管理器路径视为强约束。
• tmux 与 asciinema 的后备策略被刻意区分：tmux 调 _build_tmux_from_source()，asciinema 调 _install_asciinema_with_pip()。该差异不是由外部配置决定，而是源码中按工具名硬编码的独立恢复路径。

上下游关系

• 调用方: 未在提供源码中出现显式调用方
• 核心被调用: self.environment.exec; self._detect_system_info; self._get_combined_install_command; self._build_tmux_from_source; self._install_asciinema_with_pip; self._logger.debug; self._logger.warning
• 配置/状态来源: self._remote_asciinema_recording_path：决定是否需要检查/安装 asciinema; self.environment：提供远端命令执行能力; self._logger：承载调试与告警输出; system_info['package_manager']：由 _detect_system_info() 返回并决定是否进入包管理器安装分支
• 结果去向: 远端主机上的 tmux/asciinema 安装状态; 日志输出中的 debug / warning 记录; 后续由调用方继续使用的远端运行环境可用性; 本方法自身控制流的提前返回或后备安装分支选择

远端系统类型与包管理器探测器

stage 上下文: 该条目属于 tmux 子系统内部辅助函数，职责是把远端环境压缩成一个最小 system_info 字典，供后续内部逻辑读取。它与同 stage 的 _attempt_tmux_installation、_tmux_send_keys 一类函数不同：前者负责执行安装策略，后者负责发送 tmux 命令，而本函数只做环境探测与日志记录，不改写会话状态。

这段代码在干什么

_detect_system_info 不接收除 self 外的输入，完全依赖 self.environment.exec(...) 以 user='root' 运行若干远端命令，探测操作系统类别和首个可用包管理器。函数返回一个包含 os、package_manager、update_command 三个键的字典；其中 update_command 在当前实现中始终保持 None。除了一条 self._logger.debug(...) 调试日志外，它不写入任何 self._* 状态，也不触碰题面列出的 registers。

接口 · 参数 / IO

(self) -> dict

• 参数: self: ? — 提供远端命令执行入口 self.environment.exec 与调试日志器 self._logger
• 读状态: self.environment, self._logger
• 返回: 返回 dict[str, str | None] 风格的系统信息字典，初始键为 os、package_manager、update_command；其中前两者可能被探测填充，update_command 在本函数内不被设置。
• 副作用: 以 user='root' 执行远端命令 cat /etc/os-release 2>/dev/null || echo 'not found'; 以 user='root' 执行远端命令 uname -s; 以 user='root' 依次执行远端命令 which apt-get >/dev/null 2>&1、which dnf >/dev/null 2>&1、which yum >/dev/null 2>&1、which apk >/dev/null 2>&1、which pacman >/dev/null 2>&1、which brew >/dev/null 2>&1、which pkg >/dev/null 2>&1、which zypper >/dev/null 2>&1，直到首个返回码为 0 的候选出现后停止后续探测; 调用 self._logger.debug(f"Detected system: {system_info}") 输出调试日志

执行流

1. 先构造 system_info，把 os、package_manager、update_command 全部初始化为 None，为后续逐项填充保留默认值。
2. 无条件以 user='root' 执行 cat /etc/os-release 2>/dev/null || echo 'not found' 保存到 os_release_result，再执行 uname -s 保存到 uname_result；这两次探测在源码中都发生于任何分支判断之前。
3. 定义固定优先级的 package_managers 列表：apt-get、dnf、yum、apk、pacman、brew、pkg、zypper。随后按顺序对每个候选执行 which <pm_name> >/dev/null 2>&1；一旦某次 check_result.return_code == 0，就把该名字写入 system_info['package_manager'] 并 break，因此只记录首个命中的包管理器。
4. 若 os_release_result.return_code == 0、os_release_result.stdout 非空，且输出中不含字面量 'not found'，则把其内容转成小写后按关键词分类：包含 ubuntu 或 debian 记为 debian-based，包含 fedora 或 rhel 记为 rhel-based，包含 alpine 记为 alpine，包含 arch 记为 arch。
5. 只有在上一条 /etc/os-release 分支未成立时，代码才进入 elif uname_result.return_code == 0 and uname_result.stdout，基于 uname -s 的小写输出把 darwin 映射为 macos、把 freebsd 映射为 freebsd。
6. 最后记录 Detected system: {system_info} 调试日志并返回该字典。若 /etc/os-release 存在但内容未匹配任何已编码关键词，源码不会回退去使用 uname_result，因此 system_info['os'] 会保持 None。

源码

    async def _detect_system_info(self) -> dict:
        """
        Detect the operating system and available package managers.
        """
        system_info: dict[str, str | None] = {
            "os": None,
            "package_manager": None,
            "update_command": None,
        }

        # Check for OS release files
        os_release_result = await self.environment.exec(
            command="cat /etc/os-release 2>/dev/null || echo 'not found'",
            user="root",
        )

        # Check uname for system type
        uname_result = await self.environment.exec(command="uname -s", user="root")

        # Detect package managers by checking if they exist
        package_managers = [
            "apt-get",
            "dnf",
            "yum",
            "apk",
            "pacman",
            "brew",
            "pkg",
            "zypper",
        ]

        for pm_name in package_managers:
            check_result = await self.environment.exec(
                command=f"which {pm_name} >/dev/null 2>&1", user="root"
            )
            if check_result.return_code == 0:
                system_info["package_manager"] = pm_name
                break

        # Try to determine OS from available info
        if (
            os_release_result.return_code == 0
            and os_release_result.stdout
            and "not found" not in os_release_result.stdout
        ):
            stdout_lower = os_release_result.stdout.lower()
            if "ubuntu" in stdout_lower or "debian" in stdout_lower:
                system_info["os"] = "debian-based"
            elif "fedora" in stdout_lower or "rhel" in stdout_lower:
                system_info["os"] = "rhel-based"
            elif "alpine" in stdout_lower:
                system_info["os"] = "alpine"
            elif "arch" in stdout_lower:
                system_info["os"] = "arch"
        elif uname_result.return_code == 0 and uname_result.stdout:
            stdout_lower = uname_result.stdout.lower()
            if "darwin" in stdout_lower:
                system_info["os"] = "macos"
            elif "freebsd" in stdout_lower:
                system_info["os"] = "freebsd"

        self._logger.debug(f"Detected system: {system_info}")
        return system_info

Non-obvious 设计决策

• 操作系统探测采用“优先 /etc/os-release，其次 uname -s”的分层策略，而且 uname_result 被放在 elif 中而非独立补充判断。这意味着一旦 /etc/os-release 可读且不含 'not found'，即使内容无法识别，也不会再用 uname -s 兜底；这是当前实现中一个重要且不显眼的取舍。
• 包管理器探测与操作系统分类完全解耦：package_managers 循环在 OS 判断之前就无条件执行。因此即便 system_info['os'] 最终仍为 None，system_info['package_manager'] 也可能已成功填入某个值。
• 包管理器只保留首个命中项，而不是收集全部可用工具。对应代码是按固定顺序遍历并在首个 return_code == 0 后 break；其效果是把列表顺序编码为优先级，例如同时存在多个候选时偏向前面的 apt-get、dnf 等。
• 对 /etc/os-release 的缺失没有依赖异常处理，而是通过命令末尾的 || echo 'not found' 与后续 'not found' not in os_release_result.stdout 显式识别不可用场景。这让函数在文件缺失时仍得到可解析的标准输出分支，而不是把控制流交给异常机制。
• 返回字典中预留了 update_command 键，但本函数从头到尾未对其赋新值。这表明当前实现只负责探测并返回统一形状的数据结构，而不在这里生成更新命令。

上下游关系

• 调用方: 未知；给定源码片段未展示直接调用点; 任何需要最小系统信息字典的同类内部辅助逻辑
• 核心被调用: self.environment.exec; self._logger.debug
• 配置/状态来源: self.environment：提供远端命令执行能力; self._logger：提供调试日志输出能力; 局部常量 package_managers：规定包管理器探测顺序; 远端命令返回对象的 return_code 与 stdout：驱动 OS/包管理器分类分支
• 结果去向: 函数调用方接收返回的 system_info 字典; 调试日志 Detected system: {system_info}; 返回值中的 os 字段; 返回值中的 package_manager 字段; 返回值中的 update_command 字段（当前保持 None）
• 同类 sibling: 相较于 _attempt_tmux_installation，本函数只做探测与汇总，不执行任何安装命令。; 相较于 _send_blocking_keys、_send_non_blocking_keys 等发送路径，本函数同样调用 self.environment.exec，但用途是环境识别而非 tmux 命令执行。; 与 _tmux_send_keys 这类纯字符串组装函数不同，本函数的主要产出不是命令串，而是探测结果字典。

包管理器安装命令拼装器

stage 上下文: 该单元位于 tmux 子系统内部辅助层，职责是把包管理器标识与工具名列表折叠成一条安装命令字符串。它比同 stage 中真正执行远端命令的 _attempt_tmux_installation、_send_* 系函数更底层，只做字符串选择与拼接，不执行任何命令，也不接触实例级状态。

这段代码在干什么

_get_combined_install_command 从 system_info.get("package_manager") 读取包管理器名称，并把 tools 用空格连接成包列表，返回对应的一条安装命令字符串。若 package_manager 缺失、为 falsy 值（例如空串）或不是 str，函数直接返回空字符串 ""。函数不校验 tools 内元素类型，除 " ".join(tools) 外不做额外处理，也没有任何状态写入或外部副作用。

接口 · 参数 / IO

(self, system_info: dict, tools: list[str]) -> str

• 参数: self: TmuxSession — 方法宿主；本函数体内未读取其实例属性; system_info: dict — 提供包管理器信息；仅通过 system_info.get("package_manager") 读取; tools: list[str] — 待安装工具名列表；直接经 " ".join(tools) 形成包列表字符串
• 返回: 始终返回 str：若包管理器有效且命中内置映射，则返回对应安装命令；否则返回空字符串 ""。即使 tools 为空列表，也会返回某个字符串结果。

执行流

1. 先执行 system_info.get("package_manager") 取出 package_manager，作为后续命令模板选择键。
2. 若 package_manager 为 falsy，或其类型不是 str，立即返回空字符串 ""，不再继续拼装。
3. 把 tools 通过 " ".join(tools) 合成为单个 packages 字符串；代码未对元素内容或空列表做额外校验。
4. 构造本地字典 install_commands，为 apt-get、dnf、yum、apk、pacman、brew、pkg、zypper 预置各自的安装命令模板，并把 packages 插入模板。
5. 最后执行 install_commands.get(package_manager, "")：命中时返回对应命令，未命中时返回空字符串 ""。

源码

    def _get_combined_install_command(self, system_info: dict, tools: list[str]) -> str:
        """
        Get the appropriate installation command for multiple tools based on system info.
        """
        package_manager = system_info.get("package_manager")

        if not package_manager or not isinstance(package_manager, str):
            return ""

        # Build the package list
        packages = " ".join(tools)

        # Package manager commands with non-interactive flags
        install_commands = {
            "apt-get": f"DEBIAN_FRONTEND=noninteractive apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y {packages}",
            "dnf": f"dnf install -y {packages}",
            "yum": f"yum install -y {packages}",
            "apk": f"apk add --no-cache {packages}",
            "pacman": f"pacman -S --noconfirm {packages}",
            "brew": f"brew install {packages}",
            "pkg": f"ASSUME_ALWAYS_YES=yes pkg install -y {packages}",
            "zypper": f"zypper install -y -n {packages}",
        }

        return install_commands.get(package_manager, "")

Non-obvious 设计决策

• 函数把“不存在、空值、类型不符”的 package_manager 统一归约为返回 ""，而不是抛异常；这使调用方可以把“无法生成命令”当成普通无结果分支处理。该取舍直接体现在 if not package_manager or not isinstance(package_manager, str): return ""。
• 安装命令以静态映射表 install_commands 集中声明，而非分支链逐个拼接；这样每个包管理器的非交互参数都被固定在单处，尤其便于保留细节差异，例如 apt-get 同时在 update 与 install 前重复添加 DEBIAN_FRONTEND=noninteractive，pkg 使用 ASSUME_ALWAYS_YES=yes pkg install -y，zypper 使用 zypper install -y -n。
• 对 tools 仅做 " ".join(tools)，不做转义、去重或类型过滤；这表明本函数定位是命令模板拼装，而不是输入清洗层，复杂性被有意压低在“字符串组合”这一层面。

上下游关系

• 调用方: TmuxSession._attempt_tmux_installation（phase2 purpose 已注明会调用本函数）
• 核心被调用: system_info.get("package_manager"); " ".join(tools); install_commands.get(package_manager, ""); isinstance(package_manager, str)
• 配置/状态来源: 入参 system_info 中的 package_manager 键; 入参 tools 列表内容; 函数体内硬编码的 install_commands 模板映射; package_manager 的 truthiness 与字符串类型检查结果
• 结果去向: 作为返回值交给调用方决定是否采用该安装命令; 命中映射时输出单条 shell 命令字符串; 未命中或输入无效时输出空字符串 ""; 不写入任何 self 状态，也不直接执行外部命令
• 同类 sibling: 与 _detect_system_info 形成前后衔接：后者负责探测包管理器，本函数只消费该结果做命令映射。; 与 _attempt_tmux_installation 相比，本函数不做远端检查、安装或复验，只产生命令文本。; 与 _tmux_send_keys、_send_non_blocking_keys、_send_blocking_keys 相比，本函数完全停留在字符串构造层，不涉及执行通道。

tmux 源码安装兜底构建器

stage 上下文: 该函数属于 tmux 子系统内部安装辅助路径，角色是在常规探测或包管理器安装未得到可用 tmux 时，尝试走源码编译这一更重的兜底方案。它不参与上层会话编排，只负责远端依赖安装、下载、编译、安装与末尾验证。与同 stage 的 _attempt_tmux_installation 相比，这里不再做安装策略选择，而是承接其 fallback 分支并落地执行源码构建。

这段代码在干什么

_build_tmux_from_source 不接收显式参数，主要依赖 self.environment.exec(...) 以 user="root" 在远端执行安装与构建命令，并通过 self._logger 记录结果。函数先轮流尝试若干发行版相关的构建依赖安装命令，再下载固定版本的 tmux 3.4 源码并在 /usr/local 前缀下编译安装。它返回 None，不修改任何会话字段；真实产出是远端机器上的包安装/源码构建副作用，以及成功、失败或异常时的日志记录。

接口 · 参数 / IO

(self) -> None

• 参数: self: ? — 提供远端执行接口 environment 与日志器 _logger 的会话实例
• 读状态: self.environment, self._logger
• 返回: 返回 None；真正结果是远端依赖安装、tmux 源码下载编译安装，以及相应的 debug/error 日志。
• 副作用: 以 user="root" 在远端执行多个包管理器安装命令; 在远端 /tmp 下载 tmux-3.4.tar.gz、解压、配置、编译并执行 make install; 在远端执行 tmux -V || /usr/local/bin/tmux -V 做安装验证; 写出 self._logger.debug(...) 或 self._logger.error(...) 日志

执行流

1. 进入 try 后先构造 dep_commands 列表，内含 apt、yum、dnf、apk 四组带非交互参数的构建依赖安装命令，覆盖 build-essential/Development Tools、libevent、ncurses 与 curl 等源码构建前提。
2. 按顺序遍历 dep_commands，对每条命令调用 await self.environment.exec(command=cmd, user="root")；一旦某次返回的 result.return_code == 0 就立即 break，停止继续尝试后续发行版命令。
3. 随后组装单条 build_cmd，在 /tmp 下依次完成 curl -L 下载 tmux 3.4 发布包、tar -xzf 解压、进入 tmux-3.4 目录、执行 ./configure --prefix=/usr/local、make 与 make install。
4. 使用 await self.environment.exec(command=build_cmd, user="root") 执行整段源码构建命令；该返回值赋给 result，但后续逻辑并不检查这次构建命令的 return_code。
5. 无论前一步构建命令的返回码为何，函数都会再执行一次验证：tmux -V || /usr/local/bin/tmux -V。若 verify_result.return_code == 0，记录 debug 日志说明源码构建安装成功；否则记录 error 日志说明源码安装失败。
6. 若依赖安装、构建或验证阶段抛出 Python 层异常，则由外围 except Exception as e 捕获，并记录一条包含异常对象 e 的 error 日志；异常不会继续向上传播。

源码

    async def _build_tmux_from_source(self) -> None:
        """
        Build tmux from source as a fallback option.
        """
        try:
            # Install build dependencies based on detected system - with non-interactive flags
            dep_commands = [
                "DEBIAN_FRONTEND=noninteractive apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y build-essential libevent-dev libncurses5-dev curl",
                "yum groupinstall -y 'Development Tools' && yum install -y libevent-devel ncurses-devel curl",
                "dnf groupinstall -y 'Development Tools' && dnf install -y libevent-devel ncurses-devel curl",
                "apk add --no-cache build-base libevent-dev ncurses-dev curl",
            ]

            # Try to install build dependencies
            for cmd in dep_commands:
                result = await self.environment.exec(command=cmd, user="root")
                if result.return_code == 0:
                    break

            # Download and build tmux
            build_cmd = (
                "cd /tmp && "
                "curl -L https://github.com/tmux/tmux/releases/download/3.4/tmux-3.4.tar.gz -o tmux.tar.gz && "
                "tar -xzf tmux.tar.gz && "
                "cd tmux-3.4 && "
                "./configure --prefix=/usr/local && "
                "make && "
                "make install"
            )

            result = await self.environment.exec(command=build_cmd, user="root")

            # Verify installation
            verify_result = await self.environment.exec(
                command="tmux -V || /usr/local/bin/tmux -V", user="root"
            )
            if verify_result.return_code == 0:
                self._logger.debug("tmux successfully built and installed from source")
            else:
                self._logger.error("Failed to install tmux from source")

        except Exception as e:
            self._logger.error(f"Failed to build tmux from source: {e}")

Non-obvious 设计决策

• 依赖安装采用“多发行版命令串行试探、首个成功即接受”的策略：代码只根据 return_code == 0 提前 break，并不先调用 _detect_system_info() 做强绑定判断。这种做法降低了对系统识别准确性的依赖，但也意味着前面失败的尝试会被静默吞掉，只保留后续路径继续推进。
• 源码构建目标被硬编码为 GitHub 上的 tmux 3.4 发布包，且 ./configure 固定使用 --prefix=/usr/local。这让行为稳定、可复现，但版本选择与安装前缀都不是外部配置项，升级或适配其他目录只能改代码。
• 安装是否成功并不以 build_cmd 自身的返回码作为唯一判据，而是单独再跑一次 tmux -V || /usr/local/bin/tmux -V 验证。这体现出“以最终二进制可用性为准”的判定口径；相对地，构建命令即使失败，只要验证能找到可用 tmux，函数仍会记成功日志。
• 外围使用宽泛的 except Exception 仅记录错误而不抛出，说明该函数被设计成安装 fallback 中的容错步骤，而不是失败即中断的强保证接口。代价是调用方无法从返回值或异常上区分‘验证失败’与‘执行过程中出现异常’，只能依赖日志侧信号。

上下游关系

• 调用方: TmuxSession._attempt_tmux_installation（联合包安装路径不可用、失败或复验失败时的源码编译 fallback）; tmux 安装补救分支中的内部调用点，而非公开 orchestration 流程直接入口
• 核心被调用: self.environment.exec; self._logger.debug; self._logger.error
• 配置/状态来源: self.environment：提供远端命令执行能力与 root 用户上下文; self._logger：提供成功/失败/异常日志输出; 函数内局部常量 dep_commands：定义支持的发行版依赖安装候选集; 函数内局部常量 build_cmd：定义固定的 tmux 3.4 下载与 /usr/local 安装流程
• 结果去向: 远端系统的软件环境：可能新增构建依赖包与 /usr/local 下的 tmux 安装产物; 日志系统：写入源码安装成功、验证失败或异常失败信息; 调用方 _attempt_tmux_installation：仅通过副作用与日志间接体现结果，本函数不返回状态; 后续 tmux 可用性探测/执行路径：消费这里安装出的 tmux 或 /usr/local/bin/tmux 二进制
• 同类 sibling: 相较 _attempt_tmux_installation 的策略编排职责，本函数只负责源码构建这一条具体兜底路径。; 相较 _detect_system_info 与 _get_combined_install_command 的探测/命令拼装职责，本函数直接执行高成本安装步骤，不返回结构化系统信息或命令字符串。

asciinema 的 pip 回退安装助手

stage 上下文: 该函数属于 tmux 子系统内部安装辅助路径，负责在常规包管理器安装链路之外，为 asciinema 提供一次 pip 兜底。它通常由 _attempt_tmux_installation 在系统包安装未能确保 asciinema 可用时触发。与同 stage 的 _detect_system_info、_get_combined_install_command 不同，这里不做探测或命令生成，而是直接执行一组保守的跨发行版安装尝试；与 _build_tmux_from_source 类似，它只产生远端环境副作用与日志。

这段代码在干什么

_install_asciinema_with_pip 不接收除 self 外的额外输入，依赖 self.environment.exec(...) 以 root 身份在远端先尝试装入 python3-pip，再尝试用 pip3 或 pip 安装 asciinema。一旦某条 pip 安装命令成功，它会立刻通过 asciinema --version 复验二进制是否已可执行；验证通过则记录 debug 日志并提前返回。函数本身返回 None，不修改任何 self._* 会话字段，真实产出是远端安装副作用以及失败/成功日志；全部路径失败或中途抛异常时，只记 error 日志而不向上继续抛出。

接口 · 参数 / IO

(self) -> None

• 参数: self: ? — 提供远端执行入口 self.environment 与日志器 self._logger 的会话实例
• 读状态: self.environment, self._logger
• 返回: 返回 None；真正的产出是远端执行安装/验证命令、可能成功安装 asciinema，以及写入 debug/error 日志。
• 副作用: 以 user="root" 在远端执行多条包管理器安装命令以获取 python3-pip; 以 user="root" 在远端执行 pip3 install asciinema 或 pip install asciinema; 以 user="root" 执行 asciinema --version 做安装后验证; 调用 self._logger.debug(...) 记录安装成功; 调用 self._logger.error(...) 记录安装失败或异常

执行流

1. 进入函数后先落入最外层 try，构造 pip_install_commands：按顺序覆盖 apt-get、yum、dnf、apk 四类命令，用于先为远端准备 python3-pip。
2. 按列表顺序逐条执行 pip_install_commands，每次都通过 await self.environment.exec(command=cmd, user="root") 运行；只要某条命令返回 return_code == 0 就立即 break，不继续尝试后续包管理器命令。
3. 随后构造 pip_commands = ["pip3 install asciinema", "pip install asciinema"]，作为真正安装 asciinema 的两种候选路径。
4. 再次顺序遍历 pip_commands；对每条命令，若执行结果 return_code == 0，则立刻发起一次 asciinema --version 验证调用，确认安装不仅命令执行成功，而且二进制已可见可运行。
5. 若验证命令也返回零，则写入 self._logger.debug("asciinema successfully installed using pip") 并直接 return，终止整个回退安装流程。
6. 如果两条 pip 安装命令都未能得到通过验证的结果，函数在循环结束后统一写入 self._logger.error("Failed to install asciinema using pip")。
7. 若上述任一远端执行或日志路径抛出异常，最外层 except Exception as e 会吞掉异常并记录 self._logger.error(f"Failed to install asciinema with pip: {e}")，函数随后结束。

源码

    async def _install_asciinema_with_pip(self) -> None:
        """
        Install asciinema using pip as a fallback.
        """
        try:
            # Try to install python3-pip first - with non-interactive flags
            pip_install_commands = [
                "DEBIAN_FRONTEND=noninteractive apt-get update && DEBIAN_FRONTEND=noninteractive apt-get install -y python3-pip",
                "yum install -y python3-pip",
                "dnf install -y python3-pip",
                "apk add --no-cache python3 py3-pip",
            ]

            # Try to install pip
            for cmd in pip_install_commands:
                result = await self.environment.exec(command=cmd, user="root")
                if result.return_code == 0:
                    break

            # Install asciinema using pip
            pip_commands = ["pip3 install asciinema", "pip install asciinema"]

            for cmd in pip_commands:
                result = await self.environment.exec(command=cmd, user="root")
                if result.return_code == 0:
                    # Verify installation
                    verify_result = await self.environment.exec(
                        command="asciinema --version", user="root"
                    )
                    if verify_result.return_code == 0:
                        self._logger.debug("asciinema successfully installed using pip")
                        return

            self._logger.error("Failed to install asciinema using pip")

        except Exception as e:
            self._logger.error(f"Failed to install asciinema with pip: {e}")

Non-obvious 设计决策

• 该实现把“先补齐 pip，再装 asciinema”拆成两个顺序阶段，而不是直接假设 pip 已存在；这是因为它作为 _attempt_tmux_installation 的兜底路径，需要尽量适配缺少 Python 包工具的最小系统环境。
• 对 python3-pip 的安装采用多发行版命令串行试探，并在首个 return_code == 0 后停止，体现的是宽兼容优先而非精确探测优先；相较依赖 _detect_system_info 的单一路径，这里刻意牺牲了一些优雅性来换取回退链路的生存率。
• pip 安装成功后没有把安装命令的零返回码视为最终成功，而是追加 asciinema --version 复验；这一取舍把“命令执行成功”和“工具真正可用”区分开，避免把 PATH、包装器或环境异常误判为安装完成。
• 整个函数用最外层 try/except Exception 吞掉异常并转成 error 日志，而不是继续抛出；这说明它被设计为安装尝试中的容错分支，失败应被记录但不应在本层打断更高层的恢复/收尾逻辑。

上下游关系

• 调用方: TmuxSession._attempt_tmux_installation
• 核心被调用: self.environment.exec; self._logger.debug; self._logger.error
• 配置/状态来源: self.environment：提供远端命令执行能力; self._logger：提供成功与失败日志输出; 函数内常量列表 pip_install_commands：定义跨发行版的 pip 准备命令; 函数内常量列表 pip_commands：定义 pip 安装 asciinema 的候选命令
• 结果去向: 远端系统的软件环境：可能新增 python3-pip 与 asciinema; _attempt_tmux_installation 的回退安装语义：为其提供 pip 方案结果; 日志系统：输出安装成功、失败或异常信息; 后续依赖 asciinema 的录制路径：仅在验证通过时获得可用二进制
• 同类 sibling: 相较 _detect_system_info 与 _get_combined_install_command 的探测/拼装职责，本函数直接执行固定回退命令，不依赖系统信息字典。; 相较 _build_tmux_from_source 的源码编译回退，本函数的回退对象是 asciinema，且策略是多包管理器试探后转向 pip 安装。

生成 tmux 会话启动命令的内部助手

stage 上下文: 该函数位于 tmux 子系统的内部辅助层，职责不是实际执行远端命令，而是产出启动新 tmux session 所需的完整 shell 命令字符串。它通常在会话启动流程中被上层 TmuxSession.start 一类编排逻辑调用，与 _tmux_send_keys 这类“构造 tmux 命令但不执行”的兄弟函数同属命令生成侧。相较于安装/探测类兄弟函数，它不触发远端探测或写入，只消费既有配置状态并返回字符串。

这段代码在干什么

_tmux_start_session 根据 self._extra_env、self._pane_width、self._pane_height、self._session_name 与 self._logging_path 组装一条完整的 shell 命令，用于创建一个 detached 的 tmux 会话并把 pane 输出持续导流到日志文件。函数先把额外环境变量转换成 tmux new-session -e KEY=value 片段，再连同固定的 TERM、SHELL、pane 尺寸、session 名称和 pipe-pane 子命令拼接为单一返回值。它不执行命令、不修改实例状态，真正产出是供上层启动路径提交给远端 shell 的命令字符串。

接口 · 参数 / IO

(self) -> str

• 参数: self: ? — 提供 tmux 会话启动所需的实例配置状态
• 读状态: self._extra_env, self._pane_width, self._pane_height, self._session_name, self._logging_path
• 返回: 返回一条 shell 命令字符串：先导出固定 TERM 与 SHELL，再执行 tmux new-session ... 创建 detached session，并通过 pipe-pane 将指定 session 的 pane 输出写入 self._logging_path。
• 副作用: 无直接副作用；仅构造并返回命令字符串

执行流

1. 先遍历 self._extra_env.items()，把每个额外环境变量转成 -e <shell-quoted KEY=value> 片段，并串接成 env_options 字符串，供后续 tmux new-session 复用。
2. 随后构造返回值前半段：固定加入 export TERM=xterm-256color && export SHELL=/bin/bash &&，保证启动 tmux 之前 shell 环境具备明确终端与登录 shell 默认值。
3. 接着拼出核心 tmux new-session 命令：插入前面得到的 env_options，并使用 -x {self._pane_width}、-y {self._pane_height}、-d、-s {self._session_name} 设定 pane 尺寸、后台启动和 session 名称，最终在会话内执行 'bash --login'。
4. 最后在同一返回字符串中追加 \; pipe-pane -t {self._session_name} 'cat > {self._logging_path}'，使新建 session 的 pane 输出被持续重定向到日志文件，然后将整条命令作为 str 返回。

源码

    def _tmux_start_session(self) -> str:
        # Build environment variable options for tmux new-session -e KEY=value
        env_options = "".join(
            f"-e {shlex.quote(f'{key}={value}')} "
            for key, value in self._extra_env.items()
        )

        return (
            f"export TERM=xterm-256color && "
            f"export SHELL=/bin/bash && "
            f"tmux new-session {env_options}-x {self._pane_width} -y {self._pane_height} -d -s {self._session_name} 'bash --login' \\; "
            f"pipe-pane -t {self._session_name} "
            f"'cat > {self._logging_path}'"
        )

Non-obvious 设计决策

• 额外环境变量不是通过外层 shell 的 export 批量注入，而是逐项转成 tmux new-session -e KEY=value 参数；这使注入范围明确绑定到 tmux 会话创建动作本身，而不是隐式污染后续整条 shell 链。
• 对每个 KEY=value 使用 shlex.quote(...) 包裹，是此函数中最关键的防护性取舍：它接受 self._extra_env 的原始键值并直接拼接 shell 命令，因此显式转义可以避免空格或特殊字符破坏 tmux new-session 参数边界。
• 函数强制写入 TERM=xterm-256color 与 SHELL=/bin/bash，体现的是“固定终端默认值优先于继承调用环境”的策略；这样可减少远端默认 shell / TERM 不一致造成的 tmux 行为漂移。
• 日志捕获不是留给外层单独执行，而是在返回命令里紧跟 pipe-pane；这意味着会话一创建就建立持续输出采集，避免上层若分两步执行时出现早期输出丢失窗口。

上下游关系

• 调用方: TmuxSession.start 内部的 tmux 启动流程; 远端终端环境准备阶段中需要创建新 session 的内部编排代码
• 核心被调用: shlex.quote; dict.items（遍历 self._extra_env）
• 配置/状态来源: self._extra_env 提供需要注入 tmux 的额外环境变量; self._pane_width 提供 tmux new-session -x pane 宽度; self._pane_height 提供 tmux new-session -y pane 高度; self._session_name 提供 tmux session 名称与 pipe-pane -t 目标; self._logging_path 提供 pane 输出落盘路径
• 结果去向: 返回给上层启动路径作为待执行的远端 shell 命令; 新建 tmux session 的远端执行步骤; 后续 pane 日志持久化机制（通过 pipe-pane 写入日志文件）
• 同类 sibling: _tmux_send_keys 同样只生成 tmux 命令字符串，但面向会话创建后的按键发送，而本函数面向初始 session 启动。; _attempt_tmux_installation / _detect_system_info` 负责确保 tmux 工具可用；本函数假定这些前置条件已满足，仅负责命令组装。

构造 tmux pane 抓取命令的内部助手

stage 上下文: 该条目位于 tmux 内部辅助层，职责仅限于把 pane 抓取请求编码成一条 shell 命令字符串。它不执行命令，也不接触安装、会话启动或发送按键等同 stage 其他 helper 处理的事项。与兄弟函数相比，这里只负责 capture-pane 这一条 tmux CLI 的参数拼装。

这段代码在干什么

_tmux_capture_pane 根据布尔参数 capture_entire 与实例中的 self._session_name，返回一条完整的 tmux capture-pane 命令字符串。返回值只有两种形态：默认是 tmux capture-pane -p -t <session>；当 capture_entire=True 时变为 tmux capture-pane -p -S - -t <session>。函数本身不修改任何实例状态，也没有外部副作用。

接口 · 参数 / IO

(self, capture_entire: bool = False) -> str

• 参数: capture_entire: bool — 控制是否在命令中加入 -S - 这一组额外参数
• 读状态: self._session_name
• 返回: 返回命令字符串：当 capture_entire=False 时为 tmux capture-pane -p -t <session>；当 capture_entire=True 时为 tmux capture-pane -p -S - -t <session>。

执行流

1. 先检查 capture_entire：若为真，则把 extra_args 设为 ['-S', '-']；否则设为空列表 []。
2. 构造一个命令 token 列表，固定部分为 'tmux'、'capture-pane'、'-p'、'-t' 与 self._session_name。
3. 通过 *extra_args 把可选参数列表直接展开插入 token 序列；因此有额外参数时会得到扁平序列 ... -p -S - -t ...，而不是列表嵌套。
4. 对整个 token 序列执行 ' '.join(...)，返回最终命令字符串。

源码

    def _tmux_capture_pane(self, capture_entire: bool = False) -> str:
        if capture_entire:
            extra_args = ["-S", "-"]
        else:
            extra_args = []

        return " ".join(
            [
                "tmux",
                "capture-pane",
                "-p",
                *extra_args,
                "-t",
                self._session_name,
            ]
        )

Non-obvious 设计决策

• 函数把是否追加 -S - 的分支收敛为 extra_args 这一中间列表，再统一走一次 ' '.join(...) 返回；这样避免为两种场景分别维护两套完整命令模板。
• 使用 *extra_args 而不是把 extra_args 作为单个元素放进列表，是这里的关键取舍：它保证返回结果是平铺的命令参数序列，能够精确形成 tmux capture-pane -p -S - -t <session> 这种字符串。
• 目标 pane 只通过 self._session_name 注入到 -t 后面，说明该 helper 的职责限定在命令拼装，不在此处做 session 名称校验、命令执行或结果解析。

上下游关系

• 调用方: 未知上层 tmux pane 抓取流程；本函数源码自身未展示直接调用点
• 核心被调用: str.join
• 配置/状态来源: 参数 capture_entire; 实例状态 self._session_name
• 结果去向: 返回给上层作为待执行的 shell 命令字符串; 供后续需要 tmux capture-pane CLI 文本的路径直接使用
• 同类 sibling: 同属 [subsys-tmux-internal](subsys-tmux-internal.html)，但不同于 _tmux_start_session 生成会话启动命令、_tmux_send_keys._flush 追加发送按键命令，这里只生成 pane 抓取命令。

tmux 增量终端输出提取器

stage 上下文: 该函数属于 tmux 会话内部辅助逻辑，职责是把 pane 的当前状态整理成可直接展示的文本反馈。它本身不接收业务参数，只围绕一次完整 pane 抓取、一次历史缓冲比对，以及必要时的可见屏幕回退展开。与同 stage 中返回 tmux 命令字符串的兄弟函数不同，这里直接发起异步读取并维护 _previous_buffer 这一会话内状态。

这段代码在干什么

get_incremental_output 先抓取当前 pane 的完整缓冲区，再结合 self._previous_buffer 判断是返回“新增终端输出”还是返回“当前可见屏幕”。首次调用时没有历史基线，因此只记录当前完整缓冲区并返回可见屏幕内容；后续调用则尝试通过 _find_new_content(current_buffer) 提取增量。函数返回一个带固定标题前缀的 str，并在每次调用中把 self._previous_buffer 更新为最新完整缓冲区。

接口 · 参数 / IO

(self: TmuxSession) -> str

• 参数: self: TmuxSession — 提供 pane 抓取、可见屏幕读取、增量比对能力，并保存 _previous_buffer 历史缓冲区状态
• 读状态: self._previous_buffer
• 返回: 返回格式化字符串：要么是 "New Terminal Output:\n..."，要么是 "Current Terminal Screen:\n..."。
• 副作用: 调用 self.capture_pane(capture_entire=True) 抓取当前 pane 完整缓冲区; 在非首次分支中调用 self._find_new_content(current_buffer) 尝试比对新增内容; 在首次分支和回退分支中调用 self._get_visible_screen() 读取当前可见屏幕; 写入 self._previous_buffer 为最新的 current_buffer

执行流

1. 先执行 await self.capture_pane(capture_entire=True)，取得当前完整缓冲区并保存到局部变量 current_buffer。
2. 若 self._previous_buffer is None，说明这是首次抓取：函数把 self._previous_buffer 设为 current_buffer，再调用 await self._get_visible_screen()，返回 "Current Terminal Screen:\n..."；这一分支不会调用 _find_new_content(...)。
3. 若已存在历史缓冲区，则调用 await self._find_new_content(current_buffer)，把结果保存到 new_content。
4. 随后无论比对结果如何，都会先把 self._previous_buffer 更新为当前的 current_buffer，使后续调用以本次完整缓冲区为新基线。
5. 接着先判断 new_content is not None：若条件成立，再继续判断 new_content.strip() 是否为真。
6. 当 new_content is not None 且 new_content.strip() 为真时，返回 "New Terminal Output:\n{new_content}"；当 new_content is not None 但去空白后为空时，调用 await self._get_visible_screen() 并返回 "Current Terminal Screen:\n..."。
7. 当 new_content is None 时，函数认为无法可靠确定增量内容，于是调用 await self._get_visible_screen()，返回 "Current Terminal Screen:\n..."。

源码

    async def get_incremental_output(self) -> str:
        """
        Get either new terminal output since last call, or current screen if
        unable to determine.

        This method tracks the previous buffer state and attempts to find new content
        by comparing against the current full buffer. This provides better handling for
        commands with large output that would overflow the visible screen.

        Returns:
            str: Formatted output with either "New Terminal Output:" or
                 "Current Terminal Screen:"
        """
        current_buffer = await self.capture_pane(capture_entire=True)

        # First capture - no previous state
        if self._previous_buffer is None:
            self._previous_buffer = current_buffer
            visible_screen = await self._get_visible_screen()
            return f"Current Terminal Screen:\n{visible_screen}"

        # Try to find new content
        new_content = await self._find_new_content(current_buffer)

        # Update state
        self._previous_buffer = current_buffer

        if new_content is not None:
            if new_content.strip():
                return f"New Terminal Output:\n{new_content}"
            else:
                # No new content, show current screen
                return f"Current Terminal Screen:\n{await self._get_visible_screen()}"
        else:
            # Couldn't reliably determine new content, fall back to current screen
            return f"Current Terminal Screen:\n{await self._get_visible_screen()}"

Non-obvious 设计决策

• 首次调用不尝试构造增量，而是直接回退到可见屏幕；代码通过显式检查 self._previous_buffer is None 实现这一点，避免在没有历史基线时生成缺乏依据的“新增输出”。
• 函数把“无法可靠比对”与“比对得到的新增内容仅为空白”区分开来：前者对应 new_content is None，后者对应 new_content is not None 但 not new_content.strip()；两种情况都回退到可见屏幕，但分支语义并不相同。
• 在完成 _find_new_content(...) 后立即更新 self._previous_buffer = current_buffer，即使最终走的是回退显示路径也如此；这表明该状态的职责是记录最新完整缓冲区基线，而不是记录上一次实际返回给调用方的文本。
• 返回值总是带有固定标签前缀 New Terminal Output: 或 Current Terminal Screen:，把内容类别编码进字符串本身，减少调用方再额外判断来源类型的需要。

上下游关系

• 调用方: 未在给定源码片段中出现直接调用者
• 核心被调用: self.capture_pane(capture_entire=True); self._find_new_content(current_buffer); self._get_visible_screen()
• 配置/状态来源: self._previous_buffer
• 结果去向: 返回给本函数调用方的格式化终端反馈字符串; 写回 self._previous_buffer 作为下一次调用的历史缓冲区基线
• 同类 sibling: 同属 tmux 内部辅助函数；相较于 _tmux_capture_pane 仅构造命令字符串，本函数直接执行异步读取并维护历史缓冲区状态。

tmux 缓冲片段判定辅助函数

stage 上下文: 该函数位于 tmux 会话子系统的内部辅助层，职责比 _tmux_capture_pane 更靠后：前者只生成抓屏命令，本函数则消费抓到的 current_buffer 做字符串判定。它通常在 get_incremental_output 取得完整 pane 文本后被调用，用于把当前缓冲区与上一份快照 self._previous_buffer 做一次轻量比较。与同 stage 兄弟函数相比，它不执行远端命令、不组装 tmux shell 命令，只在内存中返回一个字符串切片或 None。

这段代码在干什么

_find_new_content 接收一个 current_buffer: str，并读取 self._previous_buffer 来构造本次比对用的 pb。若 self._previous_buffer is None，代码把 pb 设为 ""；若不是 None，则对其执行 .strip()。随后仅在 pb in current_buffer 时返回 current_buffer 的一个切片：先取 current_buffer.index(pb)，但当 pb 含有换行符时又会把该索引改写为 pb.rfind("\n")，最后返回 current_buffer[idx:]；否则返回 None。函数不修改任何实例状态，也没有外部副作用。

接口 · 参数 / IO

(self, current_buffer: str) -> str | None

• 参数: current_buffer: str — 当前一次 tmux pane 抓取得到的完整缓冲区文本
• 读状态: self._previous_buffer
• 返回: 返回 str | None。当 pb in current_buffer 时返回 current_buffer[idx:]；其中 idx 先取 current_buffer.index(pb)，若 pb 含 "\n" 则改为 pb.rfind("\n")。特别地，当 self._previous_buffer is None 时，pb 会变成空串 ""，包含判断恒成立，因此返回整个 current_buffer。若 pb 不在 current_buffer 中，则返回 None。
• 副作用: 无

执行流

1. 先读取 self._previous_buffer：若其为 None，把局部变量 pb 设为 ""；否则取 self._previous_buffer.strip() 作为 pb。
2. 检查 pb in current_buffer。若不成立，函数直接返回 None。
3. 若包含关系成立，先用 current_buffer.index(pb) 计算 idx。
4. 若 pb 中含有换行符 "\n"，代码会丢弃前一步得到的 idx，改为使用 pb.rfind("\n") 的结果作为新的 idx。
5. 返回 current_buffer[idx:] 这一字符串切片；因此在 self._previous_buffer is None 时，会返回完整的 current_buffer。

源码

    async def _find_new_content(self, current_buffer: str) -> str | None:
        pb = "" if self._previous_buffer is None else self._previous_buffer.strip()
        if pb in current_buffer:
            idx = current_buffer.index(pb)
            # Find the end of the previous buffer content
            if "\n" in pb:
                idx = pb.rfind("\n")
            return current_buffer[idx:]
        return None

Non-obvious 设计决策

• 对历史缓冲区先做 None 归一化与 .strip()，使比较输入固定落在两种形式：空串或去首尾空白后的旧快照。其直接结果是 self._previous_buffer is None 不会走异常路径，而是把空串拿去参与包含判断。
• 匹配条件只使用 pb in current_buffer，而不是做差异算法或逐行比较；因此代码只区分“包含”与“不包含”两类结果，控制流非常短，但返回值精度完全受这一包含判定约束。
• 换行分支里，代码先求 current_buffer.index(pb)，随后若 "\n" in pb 又把 idx 覆盖为 pb.rfind("\n")。这意味着最终切片位置在该分支中不再来自 current_buffer 内的匹配位置，而是来自 pb 自身内部的最后一个换行下标，并被直接用于 current_buffer[idx:]。

上下游关系

• 调用方: TmuxSession.get_incremental_output
• 核心被调用: self._previous_buffer.strip; current_buffer.index; pb.rfind
• 配置/状态来源: self._previous_buffer; current_buffer
• 结果去向: 返回给 TmuxSession.get_incremental_output 用于决定本次输出文本; 在未命中包含关系时以 None 形式交回调用方处理
• 同类 sibling: 与 _tmux_capture_pane 相比，本函数不生成 tmux 命令，而是在抓取结果到手后做纯字符串处理。; 与 get_incremental_output 相比，本函数不更新 self._previous_buffer，只返回切片或 None。

tmux 可见屏幕读取包装器

stage 上下文: 该条目位于 tmux 会话子系统的内部辅助层，角色是把一次“读取当前可见屏幕”的请求收束为单独命名的方法。与同 stage 中负责拼装命令字符串或安装依赖的兄弟函数不同，这里只有一次异步委托调用，本身不承担额外状态管理。

这段代码在干什么

_get_visible_screen 是一个异步薄包装：它只做一件事——等待 self.capture_pane(capture_entire=False) 完成，并把得到的字符串原样返回。输入只有 self，输出是一个 str。函数体本身不写入任何实例字段，也没有代码中可见的额外副作用。

接口 · 参数 / IO

(self) -> str

• 参数: self: ? — 绑定的 TmuxSession 实例；通过其 capture_pane 方法取得 pane 内容
• 读状态: 不直接读取任何可见的 self._* 字段, 间接依赖 self.capture_pane 及其所绑定的会话上下文/实例状态
• 返回: 一个协程；被 await 后产出 str，即 self.capture_pane(capture_entire=False) 的返回结果。
• 副作用: 本包装函数自身无额外状态写入; 实际外部行为完全委托给 self.capture_pane(...)

执行流

1. 调用 self.capture_pane，并显式传入 capture_entire=False。
2. 等待该异步调用完成。
3. 将得到的返回值直接作为本方法的返回值交还调用方。

源码

    async def _get_visible_screen(self) -> str:
        return await self.capture_pane(capture_entire=False)

Non-obvious 设计决策

• 函数把 capture_entire 固定为 False，用一个具名方法封装这组参数选择；代码中的非显然取舍仅此一项。

上下游关系

• 调用方: 未在给定源码片段中体现；此方法面向 TmuxSession 内部其他代码提供可见屏幕读取入口
• 核心被调用: TmuxSession.capture_pane
• 配置/状态来源: self 上绑定的 capture_pane 方法; capture_pane 所依赖的会话/实例上下文（本函数未直接展开）
• 结果去向: 直接返回给本方法的调用方; 作为一次可见 pane 内容读取结果向上传递
• 同类 sibling: TmuxSession._tmux_capture_pane 返回 tmux capture-pane 命令字符串；本函数则直接异步委托给 capture_pane(...) 取结果