# Read: /home/hyc/projects/agent-sdk/packages/comate_cli/comate_cli/terminal_agent/tui_parts/render_panels.py

Lines 1-709 of 709

     1	from __future__ import annotations
     2	
     3	import logging
     4	import time
     5	from typing import Any
     6	
     7	from prompt_toolkit.utils import get_cwidth
     8	from rich.console import Console
     9	
    10	from comate_agent_sdk.agent.queue_types import MessageOrigin, QueuedMessage
    11	
    12	from comate_cli.terminal_agent.animations import _cyan_sweep_text, breathing_dot_color, breathing_dot_glyph
    13	from comate_cli.terminal_agent.figures import (
    14	    BOTTOM_LEFT_CROP,
    15	    BULLET,
    16	    BULLET_OPERATOR,
    17	    CHECK_MARK,
    18	    DOWN_ARROW,
    19	    ELLIPSIS,
    20	    FAST_PLAY_ICON,
    21	    HEAVY_HORIZONTAL,
    22	    PAUSE_ICON,
    23	    TASK_IN_PROGRESS,
    24	    TASK_PENDING,
    25	    TASK_BLOCKED,
    26	    UP_ARROW,
    27	)
    28	from comate_cli.terminal_agent.text_effects import fit_single_line
    29	
    30	console = Console()
    31	logger = logging.getLogger(__name__)
    32	
    33	
    34	class RenderPanelsMixin:
    35	    _COMPLETION_PANEL_MAX_LINES = 8
    36	
    37	    def _terminal_width(self) -> int:
    38	        if self._app is None:
    39	            return 100
    40	        try:
    41	            return max(int(self._app.output.get_size().columns), 40)
    42	        except Exception:
    43	            return 100
    44	
    45	    def _queued_human_messages(self) -> list[QueuedMessage]:
    46	        session = getattr(self, "_session", None)
    47	        peek_queue = getattr(session, "peek_queue", None)
    48	        if not callable(peek_queue):
    49	            return []
    50	        try:
    51	            snapshot = tuple(peek_queue())
    52	        except Exception:
    53	            logger.debug("failed to read queue preview snapshot", exc_info=True)
    54	            return []
    55	        return [
    56	            message
    57	            for message in snapshot
    58	            if getattr(message, "origin", None) == MessageOrigin.HUMAN
    59	        ]
    60	
    61	    def _queue_preview_max_rows(self) -> int:
    62	        return max(1, int(getattr(self, "_queued_preview_max_rows", 3)))
    63	
    64	    def _queued_message_preview(self, message: QueuedMessage) -> str:
    65	        content = message.content
    66	        if isinstance(content, str):
    67	            preview = content
    68	        elif isinstance(content, list):
    69	            pieces: list[str] = []
    70	            for part in content:
    71	                if isinstance(part, dict):
    72	                    part_type = str(part.get("type", ""))
    73	                    if part_type == "text":
    74	                        pieces.append(str(part.get("text", "")))
    75	                    elif "image" in part_type:
    76	                        pieces.append("[image]")
    77	                else:
    78	                    pieces.append(str(part))
    79	            preview = " ".join(pieces)
    80	        else:
    81	            preview = str(content)
    82	        return " ".join(preview.split()) or "[empty message]"
    83	
    84	    def _queue_text(self) -> list[tuple[str, str]]:
    85	        queued_messages = self._queued_human_messages()[: self._queue_preview_max_rows()]
    86	        if not queued_messages:
    87	            return [("", " ")]
    88	
    89	        width = self._terminal_width()
    90	        fragments: list[tuple[str, str]] = []
    91	        for idx, message in enumerate(queued_messages):
    92	            preview = self._queued_message_preview(message)
    93	            line = fit_single_line(f"     > {preview}", width)
    94	            if idx > 0:
    95	                fragments.append(("", "\n"))
    96	            fragments.append(("class:queue.item", line))
    97	        return fragments
    98	
    99	    def _queue_height(self) -> int:
   100	        queued_count = len(self._queued_human_messages())
   101	        return max(1, min(queued_count, self._queue_preview_max_rows()))
   102	
   103	    def _status_text(self) -> list[tuple[str, str]]:
   104	        width = self._terminal_width()
   105	
   106	        # 瞬态消息优先于 mode 显示
   107	        transient = self._status_bar.transient_message
   108	        if transient:
   109	            # 右栏：model | ~branch / X% left (不变)
   110	            right_text = self._status_bar.info_status_text()
   111	            if self._busy:
   112	                git_frags = []
   113	            else:
   114	                git_frags = self._status_bar.git_diff_fragments()
   115	
   116	            right_w = sum(get_cwidth(c) for c in right_text)
   117	            if git_frags:
   118	                right_w += 2 + sum(get_cwidth(c) for _, t in git_frags for c in t)
   119	
   120	            left_w = sum(get_cwidth(c) for c in transient)
   121	            padding = max(1, width - left_w - right_w - 2)
   122	
   123	            frags: list[tuple[str, str]] = [
   124	                (
   125	                    {
   126	                        "error": "class:status.transient.error",
   127	                        "warning": "class:status.transient.warning",
   128	                    }.get(self._status_bar.transient_severity, "class:status.transient"),
   129	                    transient,
   130	                ),
   131	                ("class:status", " " * padding),
   132	                ("class:status", right_text),
   133	            ]
   134	            if git_frags:
   135	                frags.append(("class:status", "  "))
   136	                frags.extend(git_frags)
   137	            return frags
   138	
   139	        # 左栏：mode 图标 + 提示文字
   140	        mode = self._status_bar.get_mode()
   141	        hint_text = "(shift+tab to cycle)"
   142	        if mode == "plan":
   143	            left_main = f"{PAUSE_ICON} Plan mode on "
   144	            left_style = "class:status.mode.plan"
   145	        else:
   146	            left_main = f"{FAST_PLAY_ICON}{FAST_PLAY_ICON} Act mode on "
   147	            left_style = "class:status.mode.act"
   148	
   149	        # 右栏：model | ~branch / X% left
   150	        right_text = self._status_bar.info_status_text()
   151	        # busy 状态下跳过 git diff 计算，避免状态栏触发阻塞式 git 调用影响输入流畅度。
   152	        if self._busy:
   153	            git_frags = []
   154	        else:
   155	            git_frags = self._status_bar.git_diff_fragments()
   156	
   157	        left_w = sum(get_cwidth(c) for c in left_main + hint_text)
   158	        right_w = sum(get_cwidth(c) for c in right_text)
   159	        if git_frags:
   160	            right_w += 2 + sum(get_cwidth(c) for _, t in git_frags for c in t)
   161	
   162	        padding = max(1, width - left_w - right_w - 2)
   163	
   164	        frags: list[tuple[str, str]] = [
   165	            (left_style, left_main),
   166	            ("class:status.hint", hint_text),
   167	            ("class:status", " " * padding),
   168	            ("class:status", right_text),
   169	        ]
   170	        if git_frags:
   171	            frags.append(("class:status", "  "))
   172	            frags.extend(git_frags)
   173	        return frags
   174	
   175	    def _completion_panel_state(self) -> tuple[list[Any], int] | None:
   176	        input_area = getattr(self, "_input_area", None)
   177	        if input_area is None:
   178	            return None
   179	        buffer = getattr(input_area, "buffer", None)
   180	        if buffer is None:
   181	            return None
   182	        complete_state = getattr(buffer, "complete_state", None)
   183	        if complete_state is None:
   184	            return None
   185	
   186	        completions = list(getattr(complete_state, "completions", []) or [])
   187	        if not completions:
   188	            return None
   189	
   190	        selected_index: int | None = None
   191	        current_completion = getattr(complete_state, "current_completion", None)
   192	        if current_completion is not None:
   193	            for idx, completion in enumerate(completions):
   194	                if completion is current_completion:
   195	                    selected_index = idx
   196	                    break
   197	        if selected_index is None:
   198	            raw_index = getattr(complete_state, "complete_index", None)
   199	            if isinstance(raw_index, int) and 0 <= raw_index < len(completions):
   200	                selected_index = raw_index
   201	        if selected_index is None:
   202	            selected_index = 0
   203	        return completions, selected_index
   204	
   205	    def _completion_panel_text(self) -> list[tuple[str, str]]:
   206	        state = self._completion_panel_state()
   207	        if state is None:
   208	            return self._status_text()
   209	
   210	        completions, selected_index = state
   211	        width = self._terminal_width()
   212	        max_lines = max(1, int(self._COMPLETION_PANEL_MAX_LINES))
   213	        visible_slots = max_lines
   214	        hidden_below_count = 0
   215	        start = 0
   216	        end = len(completions)
   217	        if len(completions) > max_lines:
   218	            visible_slots = max(1, max_lines - 1)
   219	            start = max(0, selected_index - (visible_slots // 2))
   220	            end = min(len(completions), start + visible_slots)
   221	            start = max(0, end - visible_slots)
   222	            hidden_below_count = len(completions) - end
   223	
   224	        fragments: list[tuple[str, str]] = []
   225	        visible = completions[start:end]
   226	        slash_command_width = 0
   227	        for completion in completions:
   228	            display_text = str(getattr(completion, "display_text", completion.text))
   229	            meta_text = str(getattr(completion, "display_meta_text", "")).strip()
   230	            if meta_text and display_text.startswith("/"):
   231	                slash_command_width = max(slash_command_width, get_cwidth(display_text))
   232	
   233	        for idx, completion in enumerate(visible):
   234	            absolute_idx = start + idx
   235	            is_selected = absolute_idx == selected_index
   236	            line_style = (
   237	                "class:completion.status.current"
   238	                if is_selected
   239	                else "class:completion.status.item"
   240	            )
   241	
   242	            display_text = str(getattr(completion, "display_text", completion.text))
   243	            meta_text = str(getattr(completion, "display_meta_text", "")).strip()
   244	            line_prefix = "› " if is_selected else "  "
   245	            row_text = f"{line_prefix}{display_text}"
   246	            meta_start_index: int | None = None
   247	            render_meta_with_split_style = False
   248	            if meta_text:
   249	                if display_text.startswith("/") and slash_command_width > 0:
   250	                    command_width = get_cwidth(display_text)
   251	                    align_padding = " " * (
   252	                        max(0, slash_command_width - command_width) + 5
   253	                    )
   254	                    row_text = f"{row_text}{align_padding}{meta_text}"
   255	                    meta_start_index = (
   256	                        len(line_prefix) + len(display_text) + len(align_padding)
   257	                    )
   258	                    render_meta_with_split_style = True
   259	                else:
   260	                    row_text = f"{row_text}  —  {meta_text}"
   261	
   262	            clipped = fit_single_line(row_text, width)
   263	            padding = " " * max(0, width - get_cwidth(clipped))
   264	            if (
   265	                render_meta_with_split_style
   266	                and meta_start_index is not None
   267	                and meta_start_index < len(clipped)
   268	            ):
   269	                command_part = clipped[:meta_start_index]
   270	                meta_part = clipped[meta_start_index:]
   271	                if command_part:
   272	                    fragments.append((line_style, command_part))
   273	                if meta_part:
   274	                    fragments.append(("class:completion.status.meta", meta_part))
   275	                if padding:
   276	                    fragments.append((line_style, padding))
   277	            else:
   278	                fragments.append((line_style, f"{clipped}{padding}"))
   279	            if idx != len(visible) - 1 or hidden_below_count > 0:
   280	                fragments.append(("", "\n"))
   281	
   282	        if hidden_below_count > 0:
   283	            more_text = fit_single_line(f"{ELLIPSIS} and {hidden_below_count} more", width)
   284	            padding = " " * max(0, width - get_cwidth(more_text))
   285	            fragments.append(("class:completion.status.more", f"{more_text}{padding}"))
   286	
   287	        return fragments if fragments else [("", " ")]
   288	
   289	    def _completion_panel_height(self) -> int:
   290	        state = self._completion_panel_state()
   291	        if state is None:
   292	            return 1
   293	        completions, _ = state
   294	        max_lines = max(1, int(self._COMPLETION_PANEL_MAX_LINES))
   295	        if len(completions) <= max_lines:
   296	            return max(1, len(completions))
   297	        return max_lines
   298	
   299	    def _should_hide_loading_panel(self) -> bool:
   300	        """统一工具/loading 面板显隐逻辑。
   301	
   302	        仅当 Task 面板存在且当前没有 running tools 时隐藏该区域；一旦有
   303	        running tools，就必须保留工具面板来承载 subagent nested tools
   304	        等多行执行信息。
   305	        """
   306	        if self._renderer.has_running_tools():
   307	            return False
   308	        return self._renderer.has_active_todos()
   309	
   310	    def _sync_todo_panel_state(self) -> None:
   311	        if self._renderer.has_active_todos():
   312	            return
   313	        self._todo_panel_expanded = False
   314	        self._todo_panel_scroll = 0
   315	
   316	    def _todo_panel_view(self) -> list[str]:
   317	        self._sync_todo_panel_state()
   318	        all_lines_fn = getattr(self._renderer, "todo_all_lines", None)
   319	        if callable(all_lines_fn):
   320	            all_lines = list(all_lines_fn())
   321	        else:
   322	            all_lines = list(
   323	                self._renderer.todo_panel_lines(max_lines=self._todo_panel_max_lines)
   324	            )
   325	        if not all_lines:
   326	            return []
   327	
   328	        if not getattr(self, "_todo_panel_expanded", False):
   329	            collapsed_lines = list(
   330	                self._renderer.todo_panel_lines(max_lines=self._todo_panel_max_lines)
   331	            )
   332	            if len(all_lines) > len(collapsed_lines) and collapsed_lines:
   333	                last_line = collapsed_lines[-1]
   334	                if last_line.startswith(f"{ELLIPSIS} (+"):
   335	                    collapsed_lines[-1] = f"{last_line}, Ctrl+Y to expand"
   336	            return collapsed_lines
   337	
   338	        if len(all_lines) == 1:
   339	            return all_lines
   340	
   341	        viewport = max(2, int(self._todo_panel_expanded_max_lines))
   342	        visible_item_slots = max(1, viewport - 1)
   343	        total_items = len(all_lines) - 1
   344	        max_scroll = max(0, total_items - visible_item_slots)
   345	        offset = min(max(int(self._todo_panel_scroll), 0), max_scroll)
   346	        self._todo_panel_scroll = offset
   347	
   348	        item_lines = all_lines[1:]
   349	        visible_items = item_lines[offset : offset + visible_item_slots]
   350	        start = offset + 1
   351	        end = offset + len(visible_items)
   352	        header = (
   353	            f"{all_lines[0]} [{start}-{end}/{total_items}] "
   354	            f"Ctrl+Y collapse, {UP_ARROW}{DOWN_ARROW} scroll"
   355	        )
   356	        return [header, *visible_items]
   357	
   358	    def _loading_text(self) -> list[tuple[str, str]]:
   359	        if self._should_hide_loading_panel():
   360	            return [("", " ")]
   361	
   362	        # Running tools: multi-line tool panel with breathing dot.
   363	        if self._renderer.has_running_tools():
   364	            width = self._terminal_width()
   365	            show_flash_line = (
   366	                self._tool_result_flash_active and self._tool_panel_max_lines >= 2
   367	            )
   368	            # 动态计算所需行数，但至少保留配置值作为最小值
   369	            required_lines = self._renderer.compute_required_tool_panel_lines()
   370	            base_max = self._tool_panel_max_lines
   371	            tool_max = max(required_lines, base_max)
   372	            if show_flash_line:
   373	                tool_max = max(tool_max + 1, self._tool_panel_max_lines)
   374	            entries = self._renderer.tool_panel_entries(max_lines=max(1, tool_max))
   375	            if not entries:
   376	                return [("", " ")]
   377	
   378	            dot_color = breathing_dot_color(self._loading_frame)
   379	            now_monotonic = time.monotonic()
   380	            dot_glyph = breathing_dot_glyph(now_monotonic)
   381	            dot_style = f"fg:{dot_color} bold"
   382	            primary_style = "fg:#D1D5DB"
   383	            nested_style = "fg:#9CA3AF"
   384	
   385	            fragments: list[tuple[str, str]] = []
   386	            if show_flash_line and self._tool_result_animator.is_active:
   387	                renderable = self._tool_result_animator.renderable()
   388	                fragments.extend(self._rich_text_to_pt_fragments(renderable))
   389	                fragments.append(("", "\n"))
   390	
   391	            dim_style = "fg:#6B7280"
   392	            last_index = len(entries) - 1
   393	            for idx, (indent, line) in enumerate(entries):
   394	                if indent < 0:
   395	                    clipped = fit_single_line(line, width - 1)
   396	                    fragments.append((nested_style, clipped))
   397	                elif indent == 0:
   398	                    if isinstance(line, list):
   399	                        # Styled fragments from tool_panel_entries
   400	                        fragments.append((dot_style, f"{dot_glyph} "))
   401	                        for frag_style, frag_text in line:
   402	                            resolved = dim_style if frag_style == "class:dim" else (frag_style or primary_style)
   403	                            fragments.append((resolved, frag_text))
   404	                    else:
   405	                        clipped = fit_single_line(line, max(width - 2, 8))
   406	                        fragments.append((dot_style, f"{dot_glyph} "))
   407	                        fragments.append((primary_style, clipped))
   408	                else:
   409	                    padding = "  " * indent
   410	                    clipped = fit_single_line(line, max(width - get_cwidth(padding), 8))
   411	                    fragments.append((nested_style, padding))
   412	                    fragments.append((nested_style, clipped))
   413	
   414	                if idx != last_index:
   415	                    fragments.append(("", "\n"))
   416	
   417	            return fragments
   418	
   419	        # 优先使用动画器的渲染（流光走字 + 随机 geek 术语）
   420	        if self._tool_result_animator.is_active:
   421	            renderable = self._tool_result_animator.renderable()
   422	            frags = self._rich_text_to_pt_fragments(renderable)
   423	            return self._append_run_elapsed_fragment(frags)
   424	
   425	        if self._animator.is_active:
   426	            renderable = self._animator.renderable()
   427	            frags = self._rich_text_to_pt_fragments(renderable)
   428	            return self._append_loading_tip_fragment(
   429	                self._append_run_elapsed_fragment(frags)
   430	            )
   431	
   432	        # 获取语义化的 loading 状态
   433	        loading_state = self._renderer.loading_state()
   434	        phrase = loading_state.text.strip()
   435	        aux = self._renderer.loading_aux_text()
   436	
   437	        if not phrase:
   438	            if self._busy:
   439	                phrase = self._fallback_loading_phrase
   440	            else:
   441	                return [("", " ")]
   442	
   443	        # spec §6.7 / §7.1：容器缓冲期把 aux 拼到 phrase 后（用 dim 分隔符）
   444	        if aux:
   445	            phrase = f"{phrase}  {BULLET_OPERATOR} {aux}"
   446	
   447	        return self._append_loading_tip_fragment(
   448	            self._animated_loading_fragments(phrase)
   449	        )
   450	
   451	    def _visible_loading_tip(self) -> str:
   452	        tip = str(getattr(self, "_loading_tip_text", "") or "").strip()
   453	        if not tip:
   454	            return ""
   455	        show_at = getattr(self, "_loading_tip_show_at_monotonic", None)
   456	        if show_at is None or time.monotonic() < float(show_at):
   457	            return ""
   458	        return tip
   459	
   460	    def _append_loading_tip_fragment(
   461	        self,
   462	        frags: list[tuple[str, str]],
   463	    ) -> list[tuple[str, str]]:
   464	        tip = self._visible_loading_tip()
   465	        if not tip:
   466	            return frags
   467	
   468	        width = self._terminal_width()
   469	        prefix = f"  {BOTTOM_LEFT_CROP} "
   470	        label = " Tip: "
   471	        budget = max(width - get_cwidth(prefix + label), 8)
   472	        clipped = fit_single_line(tip, budget)
   473	        return (
   474	            frags
   475	            + [("", "\n")]
   476	            + [("fg:#555555", prefix), ("fg:#6B7280", f"{label}{clipped}")]
   477	        )
   478	
   479	    def _loading_height_with_tip(self) -> int:
   480	        return 2 if self._visible_loading_tip() else 1
   481	
   482	    def _animated_loading_fragments(self, phrase: str) -> list[tuple[str, str]]:
   483	        """用与工具调用一致的呼吸圆点 + 流光走字渲染 loading 文案."""
   484	        width = self._terminal_width()
   485	
   486	        # 若正在计时，预先计算时间后缀宽度，给 phrase 预留空间
   487	        run_start: float | None = getattr(self, "_run_start_time", None)
   488	        elapsed_suffix = ""
   489	        if run_start is not None:
   490	            elapsed = time.monotonic() - run_start
   491	            elapsed_suffix = (
   492	                f"  ({self._format_run_elapsed(elapsed)} • ctrl+c to interrupt)"
   493	            )
   494	
   495	        # glyph(1) + space(1) = 2，再减去时间后缀宽度
   496	        phrase_budget = width - 4 - get_cwidth(elapsed_suffix)
   497	        clipped = fit_single_line(phrase, max(phrase_budget, 8))
   498	
   499	        frame = self._loading_frame
   500	        dot_color = breathing_dot_color(frame)
   501	        now_monotonic = time.monotonic()
   502	
   503	        # 构建 Rich Text（与 SubmissionAnimator.renderable 相同结构）
   504	        from rich.text import Text as RichText
   505	
   506	        dot = RichText(f"{breathing_dot_glyph(now_monotonic)} ", style=f"bold {dot_color}")
   507	        sweep = _cyan_sweep_text(clipped, frame=frame)
   508	        combined = RichText.assemble(dot, sweep)
   509	        frags = self._rich_text_to_pt_fragments(combined)
   510	        if elapsed_suffix:
   511	            frags = frags + [("fg:#6B7280", elapsed_suffix)]
   512	        return frags
   513	
   514	    def _append_run_elapsed_fragment(
   515	        self,
   516	        frags: list[tuple[str, str]],
   517	    ) -> list[tuple[str, str]]:
   518	        """若当前正在计时，在 fragments 末尾追加 '  ·  Xs' 时间后缀."""
   519	        run_start: float | None = getattr(self, "_run_start_time", None)
   520	        if run_start is None:
   521	            return frags
   522	        elapsed = time.monotonic() - run_start
   523	        duration_str = self._format_run_elapsed(elapsed)
   524	        # Rich __rich_console__ 末尾会输出一个 "\n" segment，需要先剥掉再追加，
   525	        # 否则时间后缀会落在换行符之后，在单行 Window 里不可见。
   526	        tail: list[tuple[str, str]] = []
   527	        trimmed = list(frags)
   528	        while trimmed and trimmed[-1][1] == "\n":
   529	            tail.insert(0, trimmed.pop())
   530	        return (
   531	            trimmed
   532	            + [("fg:#6B7280", f"  ({duration_str} • ctrl+c to interrupt)")]
   533	            + tail
   534	        )
   535	
   536	    def _loading_height(self) -> int:
   537	        if self._should_hide_loading_panel():
   538	            return 1
   539	
   540	        # 工具面板优先：即使 animator 仍活跃，也需要为多行工具面板分配正确高度，
   541	        # 否则在 Task subagent 执行期间嵌套工具行会被 Window 裁剪为 1 行。
   542	        if self._renderer.has_running_tools():
   543	            show_flash_line = (
   544	                self._tool_result_flash_active and self._tool_panel_max_lines >= 2
   545	            )
   546	            # 动态计算所需行数，但至少保留配置值作为最小值
   547	            required_lines = self._renderer.compute_required_tool_panel_lines()
   548	            base_max = self._tool_panel_max_lines
   549	            tool_max = max(required_lines, base_max)
   550	            if show_flash_line:
   551	                tool_max = max(tool_max + 1, self._tool_panel_max_lines)
   552	            tool_lines = self._renderer.tool_panel_entries(max_lines=max(1, tool_max))
   553	            total = len(tool_lines) + (1 if show_flash_line else 0)
   554	            return max(1, total)
   555	        if self._animator.is_active:
   556	            return self._loading_height_with_tip()
   557	        if self._tool_result_animator.is_active:
   558	            return 1
   559	        if self._renderer.loading_state().text.strip():
   560	            return self._loading_height_with_tip()
   561	        if getattr(self, "_busy", False):
   562	            return self._loading_height_with_tip()
   563	        return 1
   564	
   565	    def _todo_title_shimmer_fragments(self, title: str) -> list[tuple[str, str]]:
   566	        """Todo 标题流光走字：文字不位移，仅高亮带扫过。"""
   567	        if not title:
   568	            return [("fg:#7DD3FC bold", title)]
   569	
   570	        sweep_center = (self._loading_frame % (len(title) + 8)) - 4
   571	        fragments: list[tuple[str, str]] = []
   572	        for idx, char in enumerate(title):
   573	            distance = abs(idx - sweep_center)
   574	            if distance <= 1:
   575	                style = "fg:#F8FAFC bold"
   576	            elif distance <= 3:
   577	                style = "fg:#CFFAFE bold"
   578	            elif distance <= 5:
   579	                style = "fg:#A7F3D0 bold"
   580	            else:
   581	                style = "fg:#7DD3FC bold"
   582	            fragments.append((style, char))
   583	        return fragments
   584	
   585	    def _todo_text(self) -> list[tuple[str, str]]:
   586	        lines = self._todo_panel_view()
   587	        if not lines:
   588	            return [("", " ")]
   589	
   590	        width = self._terminal_width()
   591	        fragments: list[tuple[str, str]] = []
   592	        last_index = len(lines) - 1
   593	        for idx, line in enumerate(lines):
   594	            if idx == 0:
   595	                title_elapsed_suffix = ""
   596	                run_start = getattr(self, "_run_start_time", None)
   597	                if run_start is not None:
   598	                    elapsed = time.monotonic() - run_start
   599	                    title_elapsed_suffix = (
   600	                        f"  ({self._format_run_elapsed(elapsed)} {BULLET} ctrl+c to interrupt)"
   601	                    )
   602	                # busy 时在 title 前加呼吸圆点，承载 loading 语义
   603	                dot_prefix_width = 0
   604	                if self._busy:
   605	                    dot_color = breathing_dot_color(self._loading_frame)
   606	                    dot_glyph = breathing_dot_glyph(time.monotonic())
   607	                    dot_style = f"fg:{dot_color} bold"
   608	                    fragments.append((dot_style, f"{dot_glyph} "))
   609	                    dot_prefix_width = 2  # glyph(1) + space(1)
   610	                title_budget = width - 1 - dot_prefix_width - get_cwidth(title_elapsed_suffix)
   611	                clipped_title = fit_single_line(line, max(title_budget, 8))
   612	                fragments.extend(self._todo_title_shimmer_fragments(clipped_title))
   613	                if title_elapsed_suffix:
   614	                    fragments.append(("fg:#6B7280", title_elapsed_suffix))
   615	            else:
   616	                clipped = fit_single_line(line, width - 3)  # 留空间给 prefix + space
   617	                # prefix 列：第一条任务行用 ⎿，后续用空格
   618	                is_first_task_row = idx == 1
   619	                prefix_char = BOTTOM_LEFT_CROP if is_first_task_row else " "
   620	                prefix_style = "fg:#555555"
   621	
   622	                # 解析符号和文本，应用分色
   623	                if clipped.startswith(f"{CHECK_MARK} "):
   624	                    symbol, text = CHECK_MARK, clipped[2:]
   625	                    symbol_style = "fg:#86EFAC"
   626	                    text_style = "fg:#4B5563 strike"
   627	                elif clipped.startswith(f"{TASK_IN_PROGRESS} "):
   628	                    symbol, text = TASK_IN_PROGRESS, clipped[2:]
   629	                    symbol_style = "fg:#F97316"
   630	                    text_style = "fg:#E2E8F0"
   631	                elif clipped.startswith(f"{TASK_BLOCKED} "):
   632	                    # blocked pending: TASK_BLOCKED 是中间符号，渲染时显示为 TASK_PENDING glyph
   633	                    symbol, text = TASK_PENDING, clipped[2:]
   634	                    symbol_style = "fg:#4B5563"
   635	                    text_style = "fg:#4B5563"
   636	                elif clipped.startswith(f"{TASK_PENDING} "):
   637	                    symbol, text = TASK_PENDING, clipped[2:]
   638	                    symbol_style = "fg:#6B7280"
   639	                    text_style = "fg:#6B7280"
   640	                else:
   641	                    # 溢出行 "… (+N)" 等
   642	                    symbol, text = "", clipped
   643	                    symbol_style = ""
   644	                    text_style = "fg:#6B7280"
   645	
   646	                fragments.append((prefix_style, prefix_char))
   647	                fragments.append(("", " "))
   648	                if symbol:
   649	                    fragments.append((symbol_style, symbol))
   650	                    fragments.append(("", " "))
   651	                fragments.append((text_style, text))
   652	            if idx != last_index:
   653	                fragments.append(("", "\n"))
   654	        return fragments
   655	
   656	    def _todo_height(self) -> int:
   657	        lines = self._todo_panel_view()
   658	        return max(1, len(lines))
   659	
   660	    def _rich_text_to_pt_fragments(self, renderable: Any) -> list[tuple[str, str]]:
   661	        """将 Rich Text 转换为 prompt_toolkit 的 fragments 格式."""
   662	        from rich.segment import Segment
   663	
   664	        fragments: list[tuple[str, str]] = []
   665	        for segment in renderable.__rich_console__(console, console.options):
   666	            if isinstance(segment, Segment):
   667	                text = segment.text
   668	                style = segment.style
   669	                if style:
   670	                    # 将 Rich style 转换为 prompt_toolkit style
   671	                    pt_style = self._rich_style_to_pt(style)
   672	                    fragments.append((pt_style, text))
   673	                else:
   674	                    fragments.append(("", text))
   675	        while fragments and fragments[-1][1] == "\n":
   676	            fragments.pop()
   677	        return fragments if fragments else [("", " ")]
   678	
   679	    def _rich_style_to_pt(self, rich_style: Any) -> str:
   680	        """将 Rich style 转换为 prompt_toolkit style 字符串."""
   681	        parts: list[str] = []
   682	        if rich_style.bold:
   683	            parts.append("bold")
   684	        if rich_style.italic:
   685	            parts.append("italic")
   686	        if rich_style.underline:
   687	            parts.append("underline")
   688	        if rich_style.strike:
   689	            parts.append("strike")
   690	        if rich_style.dim:
   691	            parts.append("dim")
   692	
   693	        # 前景色
   694	        if rich_style.color and rich_style.color.triplet:
   695	            parts.append(f"fg:{rich_style.color.triplet.hex}")
   696	        elif rich_style.color and rich_style.color.type.name == "STANDARD":
   697	            parts.append(f"fg:ansi{rich_style.color.number}")
   698	        elif rich_style.color and rich_style.color.type.name == "EIGHT_BIT":
   699	            parts.append(f"fg:ansi{rich_style.color.number}")
   700	
   701	        # 背景色
   702	        if rich_style.bgcolor and rich_style.bgcolor.triplet:
   703	            parts.append(f"bg:{rich_style.bgcolor.triplet.hex}")
   704	        elif rich_style.bgcolor and rich_style.bgcolor.type.name == "STANDARD":
   705	            parts.append(f"bg:ansi{rich_style.bgcolor.number}")
   706	        elif rich_style.bgcolor and rich_style.bgcolor.type.name == "EIGHT_BIT":
   707	            parts.append(f"bg:ansi{rich_style.bgcolor.number}")
   708	
   709	        return " ".join(parts) if parts else ""