sorrow404null
a7412824e0
Import the pre-repair source tree as the history baseline. Runtime data (data/), virtualenvs, bytecode caches and logs are gitignored so local secrets and user state stay out of the repo.
945 lines
38 KiB
Python
945 lines
38 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""右侧任务面板：本次会话的 bash 任务监控（运行中 / 已完成）
|
||
|
||
统一口径（与用户确认）：
|
||
· 两栏 = 「运行中」/「已完成」（按状态分）；两栏【同时可见、不互斥】，
|
||
中间是可拖动分隔线（260px 宽若真做左右并排，每栏仅 ~130px，
|
||
展开区放不下「参数 + 输出」两块列表，故用上下两栏）。
|
||
若要严格左右并排：把 _SPLIT_ORIENTATION 改成 Horizontal 即可。
|
||
· 一次 bash = 一层（BashLayer），自上而下按执行先后排列
|
||
· 单击层头 = 展开/收起；展开后分「参数」「输出」两块
|
||
· 运行中：输出为 tool_bash 的实时流（缓冲上限 200KB，超出丢弃最旧并标注）
|
||
· 已完成：输出为「进入上下文的原文」（DB messages.timeline[].result，不截断）
|
||
· 范围永远是【当前会话】；切换会话跟随刷新
|
||
· 面板【不自动收缩、也不自动展开】，开关就在面板自身：
|
||
收起态 = 栏正中按钮；展开态 = 标题行右上角按钮（永远只有一个可动按钮）
|
||
· 🆕 左边缘可拖拽调宽（最小 200px）；松手立即记录到 config.json["bash_panel_width"]，
|
||
下次按展开按钮自动恢复到记录的宽度
|
||
· 「已出上下文」= 该层在最近一次压缩切点之前（其输出已不在 API 上下文里）
|
||
"""
|
||
import json
|
||
import os
|
||
import re
|
||
|
||
from PyQt6 import QtCore, QtGui, QtWidgets
|
||
|
||
LIVE_BUF_CAP = 200 * 1024 # 每条实时输出缓冲上限
|
||
LAYER_LIMIT = 30 # 已完成栏默认只渲染最近 N 层
|
||
_SPLIT_ORIENTATION = QtCore.Qt.Orientation.Vertical # 改 Horizontal = 左右并排
|
||
|
||
_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
_DUR_RE = re.compile(r"\((\d+(?:\.\d+)?)s\)\s*$")
|
||
|
||
# ---------------------------------------------------------------- 宽度
|
||
PANEL_W_DEFAULT = 260 # 首次运行的默认展开宽度
|
||
PANEL_W_MIN = 200 # 🆕 拖拽最小宽度（再窄「参数/输出」两块就挤坏了）
|
||
PANEL_W_MAX = 560 # 绝对上限（实际还会受「不超过主窗口 50%」约束）
|
||
_CFG_KEY = "bash_panel_width"
|
||
_CFG_PATH = os.path.join(_ROOT, "data", "config.json")
|
||
|
||
|
||
def _cfg_path() -> str:
|
||
"""配置文件路径（测试可用 HAOCODE_CONFIG_FILE 指向临时文件 → 绝不污染真配置）"""
|
||
return os.environ.get("HAOCODE_CONFIG_FILE") or _CFG_PATH
|
||
|
||
|
||
def load_panel_width(default: int = PANEL_W_DEFAULT) -> int:
|
||
"""读回上次拖拽记录的宽度；缺失/异常 → 默认值（并强制夹在 [MIN, MAX]）"""
|
||
try:
|
||
with open(_cfg_path(), "r", encoding="utf-8") as f:
|
||
w = int(json.load(f).get(_CFG_KEY, default))
|
||
except Exception:
|
||
return default
|
||
return max(PANEL_W_MIN, min(PANEL_W_MAX, w))
|
||
|
||
|
||
def save_panel_width(w: int) -> bool:
|
||
"""只改这一个键、其余配置原样保留；临时文件 + replace 原子写"""
|
||
p = _cfg_path()
|
||
try:
|
||
cfg = {}
|
||
if os.path.exists(p):
|
||
with open(p, "r", encoding="utf-8") as f:
|
||
cfg = json.load(f) or {}
|
||
cfg[_CFG_KEY] = int(w)
|
||
tmp = p + ".tmp"
|
||
with open(tmp, "w", encoding="utf-8") as f:
|
||
json.dump(cfg, f, ensure_ascii=False, indent=2)
|
||
os.replace(tmp, p)
|
||
return True
|
||
except Exception as e:
|
||
print(f"[UI]: 保存任务面板宽度失败: {e}")
|
||
return False
|
||
|
||
|
||
def _svg(name: str) -> str:
|
||
return os.path.join(_ROOT, "svg", name)
|
||
|
||
|
||
def _fmt_dur(sec) -> str:
|
||
try:
|
||
sec = float(sec)
|
||
except Exception:
|
||
return ""
|
||
return f"{sec:.1f}s" if sec < 60 else f"{int(sec // 60)}m{int(sec % 60)}s"
|
||
|
||
|
||
def _parse_dur(text):
|
||
"""从上下文文本结尾的 (Ns) 里取耗时；取不到返回 None"""
|
||
m = _DUR_RE.search((text or "").strip())
|
||
if not m:
|
||
return None
|
||
try:
|
||
return float(m.group(1))
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _cmd_of(args) -> str:
|
||
"""args 可能是 JSON 字符串或 dict → 取出 bash 的 command"""
|
||
if isinstance(args, dict):
|
||
return args.get("command", "") or ""
|
||
try:
|
||
d = json.loads(args or "{}")
|
||
if isinstance(d, dict):
|
||
return d.get("command", "") or ""
|
||
except Exception:
|
||
pass
|
||
return str(args or "")
|
||
|
||
|
||
def _iter_bash_entries(timeline):
|
||
"""timeline 可能是 JSON 字符串（DB）或 list[dict]（内存）→ 产出 name=='bash' 的工具条目"""
|
||
data = timeline
|
||
if isinstance(timeline, str):
|
||
if not timeline.strip():
|
||
return
|
||
try:
|
||
data = json.loads(timeline)
|
||
except Exception:
|
||
return
|
||
if not isinstance(data, list):
|
||
return
|
||
for e in data:
|
||
if isinstance(e, dict) and e.get("t") == "tool" and e.get("name") == "bash":
|
||
yield e
|
||
|
||
|
||
class _ClickFrame(QtWidgets.QFrame):
|
||
"""整块可单击（单击即触发，不用双击）"""
|
||
clicked = QtCore.pyqtSignal()
|
||
|
||
def mouseReleaseEvent(self, e):
|
||
if (e.button() == QtCore.Qt.MouseButton.LeftButton
|
||
and self.rect().contains(e.position().toPoint())):
|
||
self.clicked.emit()
|
||
super().mouseReleaseEvent(e)
|
||
|
||
|
||
# ======================================================================
|
||
# 一层 = 一次 bash 执行
|
||
# ======================================================================
|
||
class BashLayer(QtWidgets.QFrame):
|
||
def __init__(self, call_id: str, command: str = "", parent=None):
|
||
super().__init__(parent)
|
||
self.setObjectName("bl_layer")
|
||
self.call_id = call_id
|
||
self.command = command or ""
|
||
self.ok = None # None=运行中 / True / False=已完成
|
||
self.in_context = True
|
||
self.expanded = False
|
||
self.elapsed = 0
|
||
self.timeout = 0
|
||
self._live = ""
|
||
self._live_truncated = False
|
||
self._final = None
|
||
self._build()
|
||
self._apply_status()
|
||
|
||
# ---------------- UI ----------------
|
||
def _build(self):
|
||
v = QtWidgets.QVBoxLayout(self)
|
||
v.setContentsMargins(0, 0, 0, 0)
|
||
v.setSpacing(0)
|
||
|
||
self.head = _ClickFrame()
|
||
self.head.setObjectName("bl_head")
|
||
self.head.setCursor(QtCore.Qt.CursorShape.PointingHandCursor)
|
||
self.head.setToolTip(self.command or "")
|
||
hl = QtWidgets.QHBoxLayout(self.head)
|
||
hl.setContentsMargins(8, 6, 8, 6)
|
||
hl.setSpacing(6)
|
||
|
||
self.dot = QtWidgets.QLabel("●")
|
||
self.dot.setObjectName("bl_dot")
|
||
self.name = QtWidgets.QLabel("bash")
|
||
self.name.setObjectName("bl_name")
|
||
self.meta = QtWidgets.QLabel("")
|
||
self.meta.setObjectName("bl_meta")
|
||
self.tag = QtWidgets.QLabel("")
|
||
self.tag.setObjectName("bl_tag")
|
||
self.cmd = QtWidgets.QLabel(self._preview())
|
||
self.cmd.setObjectName("bl_cmd")
|
||
self.cmd.setSizePolicy(QtWidgets.QSizePolicy.Policy.Ignored,
|
||
QtWidgets.QSizePolicy.Policy.Preferred)
|
||
self.chev = QtWidgets.QLabel("▸")
|
||
self.chev.setObjectName("bl_chev")
|
||
|
||
hl.addWidget(self.dot, 0)
|
||
hl.addWidget(self.name, 0)
|
||
hl.addWidget(self.meta, 0)
|
||
hl.addWidget(self.tag, 0)
|
||
hl.addWidget(self.cmd, 1)
|
||
hl.addWidget(self.chev, 0)
|
||
v.addWidget(self.head)
|
||
|
||
self.body = QtWidgets.QWidget()
|
||
self.body.setObjectName("bl_body")
|
||
bl = QtWidgets.QVBoxLayout(self.body)
|
||
bl.setContentsMargins(8, 0, 8, 8)
|
||
bl.setSpacing(4)
|
||
|
||
self.lbl_arg = QtWidgets.QLabel("参数")
|
||
self.lbl_arg.setObjectName("bl_sect_label")
|
||
self.arg_box = self._code_box(80, wrap=True)
|
||
bl.addWidget(self.lbl_arg)
|
||
bl.addWidget(self.arg_box)
|
||
|
||
self.lbl_out = QtWidgets.QLabel("输出")
|
||
self.lbl_out.setObjectName("bl_sect_label")
|
||
self.out_box = self._code_box(230, wrap=False)
|
||
bl.addWidget(self.lbl_out)
|
||
bl.addWidget(self.out_box)
|
||
|
||
self.body.hide()
|
||
v.addWidget(self.body)
|
||
self.head.clicked.connect(self.toggle)
|
||
|
||
def _code_box(self, max_h, wrap):
|
||
box = QtWidgets.QPlainTextEdit()
|
||
box.setObjectName("bl_code")
|
||
box.setReadOnly(True)
|
||
box.setMaximumHeight(max_h)
|
||
box.setLineWrapMode(
|
||
QtWidgets.QPlainTextEdit.LineWrapMode.WidgetWidth if wrap
|
||
else QtWidgets.QPlainTextEdit.LineWrapMode.NoWrap)
|
||
box.setFrameShape(QtWidgets.QFrame.Shape.NoFrame)
|
||
box.setVerticalScrollBarPolicy(QtCore.Qt.ScrollBarPolicy.ScrollBarAsNeeded)
|
||
box.setHorizontalScrollBarPolicy(
|
||
QtCore.Qt.ScrollBarPolicy.ScrollBarAsNeeded if not wrap
|
||
else QtCore.Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
|
||
return box
|
||
|
||
def _preview(self) -> str:
|
||
s = " ".join((self.command or "").split())
|
||
return s[:58] + ("…" if len(s) > 58 else "")
|
||
|
||
# ---------------- 状态 ----------------
|
||
def _apply_status(self):
|
||
if self.ok is None:
|
||
self.dot.setProperty("state", "run")
|
||
self.meta.setText(f"{self.elapsed}/{self.timeout}s" if self.timeout
|
||
else (f"{self.elapsed}s" if self.elapsed else "…"))
|
||
else:
|
||
self.dot.setProperty("state", "ok" if self.ok else "bad")
|
||
self.meta.setText(_fmt_dur(self.elapsed) if self.elapsed else "")
|
||
|
||
if self.ok is not None and not self.in_context:
|
||
self.tag.setText("已出上下文")
|
||
self.tag.setProperty("kind", "out")
|
||
elif self.ok is None:
|
||
self.tag.setText("运行中")
|
||
self.tag.setProperty("kind", "run")
|
||
elif self.ok:
|
||
self.tag.setText("已完成")
|
||
self.tag.setProperty("kind", "ok")
|
||
else:
|
||
self.tag.setText("失败")
|
||
self.tag.setProperty("kind", "bad")
|
||
|
||
for w in (self.dot, self.tag):
|
||
w.style().unpolish(w)
|
||
w.style().polish(w)
|
||
|
||
def set_running(self, elapsed, timeout):
|
||
self.ok = None
|
||
self.elapsed = int(elapsed or 0)
|
||
self.timeout = int(timeout or 0)
|
||
self._apply_status()
|
||
if self.expanded:
|
||
self._render_output()
|
||
|
||
def append_live(self, text: str):
|
||
if not text:
|
||
return
|
||
self._live += text
|
||
if len(self._live) > LIVE_BUF_CAP:
|
||
self._live = self._live[-LIVE_BUF_CAP:]
|
||
self._live_truncated = True
|
||
if self.expanded:
|
||
self._render_output()
|
||
|
||
def set_finished(self, ok, text):
|
||
self.ok = bool(ok)
|
||
self._final = text or ""
|
||
d = _parse_dur(self._final)
|
||
if d is not None:
|
||
self.elapsed = d
|
||
self._apply_status()
|
||
if self.expanded:
|
||
self._render_output()
|
||
|
||
def set_in_context(self, in_ctx: bool):
|
||
self.in_context = bool(in_ctx)
|
||
self._apply_status()
|
||
|
||
# ---------------- 展开 ----------------
|
||
def toggle(self):
|
||
self.expanded = not self.expanded
|
||
self.body.setVisible(self.expanded)
|
||
self.chev.setText("▾" if self.expanded else "▸")
|
||
if self.expanded:
|
||
self.arg_box.setPlainText(self.command or "（无参数）")
|
||
self._render_output()
|
||
|
||
def _render_output(self):
|
||
if self.ok is None:
|
||
body = self._live
|
||
if self._live_truncated:
|
||
body = "…（实时缓冲已截断，仅保留最后 200KB）\n" + body
|
||
self.lbl_out.setText("输出（实时）")
|
||
self.out_box.setPlainText(body if body else "（等待输出…）")
|
||
sb = self.out_box.verticalScrollBar()
|
||
sb.setValue(sb.maximum())
|
||
else:
|
||
self.lbl_out.setText("输出（进入上下文）")
|
||
self.out_box.setPlainText(self._final or "（无输出）")
|
||
|
||
|
||
# ======================================================================
|
||
# 一栏：标题（可折叠）+ 层列表
|
||
# ======================================================================
|
||
class _Section(QtWidgets.QWidget):
|
||
def __init__(self, title, empty_hint, on_fold=None, parent=None):
|
||
super().__init__(parent)
|
||
self.empty_hint = empty_hint
|
||
self.folded = False
|
||
self._on_fold = on_fold # 回调：交面板做高度动画（抽屉式）
|
||
self._effect = None # 内容区淡入淡出用（按需创建，静止时移除）
|
||
self._head_hint = 0 # 头部行高缓存（恒定值）
|
||
v = QtWidgets.QVBoxLayout(self)
|
||
v.setContentsMargins(0, 0, 0, 0)
|
||
v.setSpacing(0)
|
||
|
||
self.head = _ClickFrame()
|
||
self.head.setObjectName("bl_sect_head")
|
||
self.head.setCursor(QtCore.Qt.CursorShape.PointingHandCursor)
|
||
hl = QtWidgets.QHBoxLayout(self.head)
|
||
hl.setContentsMargins(10, 6, 8, 6)
|
||
hl.setSpacing(6)
|
||
self.chev = QtWidgets.QLabel("▾")
|
||
self.chev.setObjectName("bl_sect_chev")
|
||
self.title = QtWidgets.QLabel(title)
|
||
self.title.setObjectName("bl_sect_title")
|
||
self.count = QtWidgets.QLabel("0")
|
||
self.count.setObjectName("bl_sect_count")
|
||
hl.addWidget(self.chev, 0)
|
||
hl.addWidget(self.title, 0)
|
||
hl.addStretch(1)
|
||
hl.addWidget(self.count, 0)
|
||
v.addWidget(self.head)
|
||
|
||
self.scroll = QtWidgets.QScrollArea()
|
||
self.scroll.setObjectName("bl_scroll")
|
||
self.scroll.setWidgetResizable(True)
|
||
self.scroll.setFrameShape(QtWidgets.QFrame.Shape.NoFrame)
|
||
self.scroll.setHorizontalScrollBarPolicy(QtCore.Qt.ScrollBarPolicy.ScrollBarAlwaysOff)
|
||
# 🐛 修复：QScrollArea 的 viewport 默认按 palette base（白）自画背景，
|
||
# 会把父级 #f7f8fa 盖成白色 → 强制透明
|
||
self.scroll.viewport().setStyleSheet("background: transparent;")
|
||
self.scroll.viewport().setAutoFillBackground(False)
|
||
self.host = QtWidgets.QWidget()
|
||
self.host.setObjectName("bl_host")
|
||
self.host.setAutoFillBackground(False)
|
||
self.lay = QtWidgets.QVBoxLayout(self.host)
|
||
self.lay.setContentsMargins(6, 2, 6, 6)
|
||
self.lay.setSpacing(5)
|
||
self.scroll.setWidget(self.host)
|
||
|
||
# 🆕 「下面填充的 bash 层」整块放进 body：折叠/展开动画只针对它，
|
||
# 头部行（self.head）高度永远不变
|
||
self.body = QtWidgets.QWidget()
|
||
self.body.setObjectName("bl_body")
|
||
self.body.setAutoFillBackground(False)
|
||
bv = QtWidgets.QVBoxLayout(self.body)
|
||
bv.setContentsMargins(0, 0, 0, 0)
|
||
bv.setSpacing(0)
|
||
bv.addWidget(self.scroll, 1)
|
||
|
||
self.hint = QtWidgets.QLabel("")
|
||
self.hint.setObjectName("bl_hint")
|
||
self.hint.setWordWrap(True)
|
||
bv.addWidget(self.hint)
|
||
v.addWidget(self.body, 1)
|
||
|
||
self.head.clicked.connect(self._toggle_fold)
|
||
self._set_hint(self.empty_hint)
|
||
# 🐛 头部行高【硬固定】：收起时 body 被隐、布局里只剩 head，而 QFrame 默认
|
||
# 竖直策略可伸长 → head 会被拉满整栏（实测 793px 巨条 = 「按钮变大很高」）
|
||
self._head_hint = max(16, self.head.sizeHint().height())
|
||
self.head.setFixedHeight(self._head_hint)
|
||
|
||
# ---------------- 头部行高（收起/展开恒定的那一行）----------------
|
||
def head_height(self) -> int:
|
||
# 🐛 必须缓存 sizeHint：若用实时 self.head.height()，会形成正反馈——
|
||
# 头部被压矮 → minimumSizeHint 变小 → splitter 允许更矮 → 头部再被压矮…
|
||
if self._head_hint <= 0:
|
||
self._head_hint = max(16, self.head.sizeHint().height())
|
||
return self._head_hint
|
||
def minimumSizeHint(self):
|
||
# 🐛 关键修复：默认最小高 ≈ 113px（头部 + QScrollArea 最小高 + 提示语），
|
||
# QSplitter 完全没法把本栏收缩 → 收起后留一大块空白（用户报的 bug）。
|
||
# 改为「只剩头部行」，内容区允许被压到 0（超出部分天然被父级裁切）。
|
||
return QtCore.QSize(0, self.head_height())
|
||
|
||
# ---------------- 内容区透明度（动画帧驱动）----------------
|
||
def set_body_opacity(self, op: float):
|
||
op = max(0.0, min(1.0, float(op)))
|
||
if op >= 0.999:
|
||
self._drop_effect()
|
||
return
|
||
if self._effect is None:
|
||
self._effect = QtWidgets.QGraphicsOpacityEffect(self.body)
|
||
self.body.setGraphicsEffect(self._effect)
|
||
self._effect.setOpacity(op)
|
||
|
||
def _drop_effect(self):
|
||
"""去掉离屏渲染开销（静止时不挂 effect）"""
|
||
if self.body.graphicsEffect() is not None:
|
||
self.body.setGraphicsEffect(None)
|
||
self._effect = None
|
||
|
||
def show_body(self):
|
||
self.body.setVisible(True)
|
||
self.scroll.setVisible(True)
|
||
self.hint.setVisible(bool(self.hint.text()))
|
||
|
||
def hide_body(self):
|
||
self.body.setVisible(False)
|
||
|
||
def set_folded(self, folded: bool):
|
||
"""应用折叠状态（高度由面板的 splitter 动画负责）"""
|
||
self.folded = bool(folded)
|
||
self.chev.setText("▸" if self.folded else "▾")
|
||
if self.folded:
|
||
self._drop_effect()
|
||
else:
|
||
self.show_body()
|
||
self.set_body_opacity(1.0)
|
||
|
||
def _toggle_fold(self):
|
||
if self._on_fold is not None:
|
||
self._on_fold(self) # 交面板：状态切换 + 高度抽屉动画
|
||
return
|
||
self.set_folded(not self.folded)
|
||
self.scroll.setVisible(not self.folded)
|
||
self.hint.setVisible(not self.folded and bool(self.hint.text()))
|
||
|
||
def _set_hint(self, text):
|
||
self.hint.setText(text or "")
|
||
self.hint.setVisible(not self.folded and bool(text))
|
||
|
||
def set_layers(self, layers, total=None):
|
||
"""重排层（复用同一批 widget → 实时缓冲不丢）"""
|
||
while self.lay.count():
|
||
it = self.lay.takeAt(0)
|
||
w = it.widget()
|
||
if w is not None:
|
||
w.setParent(None)
|
||
for w in layers:
|
||
self.lay.addWidget(w)
|
||
w.show()
|
||
self.lay.addStretch(1)
|
||
self.count.setText(str(len(layers)))
|
||
total = len(layers) if total is None else total
|
||
if total > len(layers):
|
||
self._set_hint(f"仅显示最近 {len(layers)} 层（本会话共 {total} 层）")
|
||
else:
|
||
self._set_hint("" if layers else self.empty_hint)
|
||
|
||
|
||
# ======================================================================
|
||
# 右侧任务面板（外壳，镜像左侧边栏）
|
||
# ======================================================================
|
||
class _ResizeHandle(QtWidgets.QWidget):
|
||
"""面板左边缘的拖拽手柄（4px 隐形条，hover 才亮；光标 SizeHor）"""
|
||
|
||
W = 4
|
||
|
||
def __init__(self, panel):
|
||
super().__init__(panel)
|
||
self._panel = panel
|
||
self.setObjectName("bl_resize_handle")
|
||
# QWidget 子类必须显式开启才会绘制样式表背景（hover 高亮靠它）
|
||
self.setAttribute(QtCore.Qt.WidgetAttribute.WA_StyledBackground, True)
|
||
self.setCursor(QtCore.Qt.CursorShape.SizeHorCursor)
|
||
self.setStyleSheet("#bl_resize_handle:hover { background: #dfe3ea; }")
|
||
self.setToolTip("拖拽调整任务面板宽度（最小 %dpx）" % PANEL_W_MIN)
|
||
self._drag = False
|
||
|
||
def mousePressEvent(self, e):
|
||
if e.button() == QtCore.Qt.MouseButton.LeftButton:
|
||
self._drag = True
|
||
self._panel._drag_begin(e.globalPosition().x())
|
||
e.accept()
|
||
return
|
||
super().mousePressEvent(e)
|
||
|
||
def mouseMoveEvent(self, e):
|
||
if self._drag:
|
||
self._panel._drag_to_global_x(e.globalPosition().x())
|
||
e.accept()
|
||
return
|
||
super().mouseMoveEvent(e)
|
||
|
||
def mouseReleaseEvent(self, e):
|
||
if self._drag:
|
||
self._drag = False
|
||
self._panel._commit_drag_width()
|
||
e.accept()
|
||
return
|
||
super().mouseReleaseEvent(e)
|
||
|
||
|
||
class BashPanel(QtWidgets.QWidget):
|
||
W_EXPAND = PANEL_W_DEFAULT
|
||
W_COLLAPSE = 52
|
||
W_MIN = PANEL_W_MIN
|
||
W_MAX = PANEL_W_MAX
|
||
FOLD_MS = 200 # 栏折叠/展开时长（与 ModelSelectPopup.DRAWER_DURATION 一致）
|
||
FOLD_STEP = 16 # 帧间隔（与 DRAWER_STEP 一致，~60fps）
|
||
MIN_BODY_H = 60 # 展开后「下面填充的层」至少保留的高度
|
||
|
||
def __init__(self, parent=None):
|
||
super().__init__(parent)
|
||
self.setObjectName("right_sidebar")
|
||
# 🐛 关键修复：QWidget 的自定义子类默认【不绘制样式表背景/边框】
|
||
# （#sidebar 是原生 QWidget 实例所以正常，BashPanel 是子类所以失效——
|
||
# 表现为右侧栏底色变白、左侧分隔线丢失、空间感消失）
|
||
self.setAttribute(QtCore.Qt.WidgetAttribute.WA_StyledBackground, True)
|
||
# 🆕 展开宽度：优先用上次拖拽记录的值（config.json: bash_panel_width）
|
||
self.W_EXPAND = load_panel_width()
|
||
self._end_w = self.W_EXPAND
|
||
self._drag_right = None
|
||
self._w = self.W_COLLAPSE
|
||
self.setFixedWidth(self._w) # 默认收起
|
||
self.collapsed = True
|
||
self._target = True
|
||
self._anim = None
|
||
self._session_id = None
|
||
self._db = None
|
||
self._layers = {} # call_id -> BashLayer
|
||
self._order = [] # 执行先后
|
||
self._running = set()
|
||
self._done = set()
|
||
self._saved_h = [120, 520] # 每栏收起前的高度（展开时恢复）
|
||
self._fold = None # 折叠动画状态
|
||
self._fold_timer = QtCore.QTimer(self)
|
||
self._fold_timer.setInterval(self.FOLD_STEP)
|
||
self._fold_timer.timeout.connect(self._fold_tick)
|
||
self._build()
|
||
# 🆕 左边缘拖拽手柄（浮在最左 4px，收起/展开两种状态都在）
|
||
self._handle = _ResizeHandle(self)
|
||
self._handle.setFixedWidth(_ResizeHandle.W)
|
||
self._handle.raise_()
|
||
|
||
# ---------------- 外壳 ----------------
|
||
def _build(self):
|
||
v = QtWidgets.QVBoxLayout(self)
|
||
v.setContentsMargins(0, 0, 0, 0)
|
||
v.setSpacing(0)
|
||
self.stack = QtWidgets.QStackedWidget()
|
||
v.addWidget(self.stack)
|
||
|
||
# ---- 展开页（顶部留白 20px + 36px 标题行，与左侧栏完全对齐）----
|
||
self.expand_page = QtWidgets.QWidget()
|
||
ev = QtWidgets.QVBoxLayout(self.expand_page)
|
||
ev.setContentsMargins(0, 20, 0, 12)
|
||
ev.setSpacing(0)
|
||
|
||
self.header = QtWidgets.QWidget()
|
||
self.header.setObjectName("bl_panel_head")
|
||
self.header.setFixedHeight(36)
|
||
hl = QtWidgets.QHBoxLayout(self.header)
|
||
hl.setContentsMargins(10, 0, 10, 0)
|
||
hl.setSpacing(6)
|
||
self.lbl_title = QtWidgets.QLabel("任务面板")
|
||
self.lbl_title.setObjectName("bl_panel_title")
|
||
self.lbl_total = QtWidgets.QLabel("")
|
||
self.lbl_total.setObjectName("bl_panel_total")
|
||
# 🐛 修复双按钮：面板内不再自带收起按钮（唯一开关 = 顶部工具栏按钮），
|
||
# 与左侧栏的区别：左侧栏的开关在栏内，右侧栏的开关在顶部（原「导出」位置）
|
||
hl.addWidget(self.lbl_title, 0)
|
||
hl.addWidget(self.lbl_total, 0)
|
||
hl.addStretch(1)
|
||
# 🆕 展开态开关：标题行右端（与左侧栏 collapse_btn 完全同规格）
|
||
self.fold_btn = QtWidgets.QPushButton("")
|
||
self.fold_btn.setObjectName("collapse_btn")
|
||
self.fold_btn.setIcon(QtGui.QIcon(_svg("panel_right.svg")))
|
||
self.fold_btn.setIconSize(QtCore.QSize(16, 16))
|
||
self.fold_btn.setFixedSize(28, 28)
|
||
self.fold_btn.setCursor(QtCore.Qt.CursorShape.PointingHandCursor)
|
||
self.fold_btn.setToolTip("收起任务面板")
|
||
hl.addWidget(self.fold_btn, 0)
|
||
ev.addWidget(self.header)
|
||
ev.addSpacing(10)
|
||
|
||
self.splitter = QtWidgets.QSplitter(_SPLIT_ORIENTATION)
|
||
self.splitter.setObjectName("bl_splitter")
|
||
self.splitter.setChildrenCollapsible(False)
|
||
self.splitter.setHandleWidth(5)
|
||
self.sec_running = _Section("运行中", "暂无正在运行的 bash", on_fold=self._toggle_section)
|
||
self.sec_done = _Section("已完成", "本会话还没有已完成的 bash", on_fold=self._toggle_section)
|
||
self.splitter.addWidget(self.sec_running)
|
||
self.splitter.addWidget(self.sec_done)
|
||
# 🆕 底部留白占位：最后一栏收起时由它吸收余量（否则栏头会被顶到面板底部，
|
||
# 或被拉伸成巨条）
|
||
self.spacer = QtWidgets.QWidget()
|
||
self.spacer.setObjectName("bl_spacer")
|
||
self.spacer.setAttribute(QtCore.Qt.WidgetAttribute.WA_StyledBackground, True)
|
||
self.spacer.setStyleSheet("#bl_spacer { background: transparent; }")
|
||
self.splitter.addWidget(self.spacer)
|
||
self.splitter.setCollapsible(2, True) # 占位层允许被压到 0
|
||
self.splitter.setSizes(self._saved_h + [0])
|
||
_h2 = self.splitter.handle(2)
|
||
if _h2 is not None: # 占位层上方的分隔条：不可拖、不可见
|
||
_h2.setEnabled(False)
|
||
_h2.setStyleSheet("background: transparent;")
|
||
ev.addWidget(self.splitter, 1)
|
||
self.stack.addWidget(self.expand_page)
|
||
|
||
# ---- 收起页：栏正中一个展开按钮（与左侧栏 collapse_expand_btn 完全同规格）----
|
||
self.collapse_page = QtWidgets.QWidget()
|
||
cl = QtWidgets.QVBoxLayout(self.collapse_page)
|
||
cl.setContentsMargins(0, 0, 0, 0)
|
||
cl.addStretch(1)
|
||
self.expand_btn = QtWidgets.QPushButton("")
|
||
self.expand_btn.setObjectName("collapse_btn")
|
||
self.expand_btn.setIcon(QtGui.QIcon(_svg("panel_right.svg")))
|
||
self.expand_btn.setIconSize(QtCore.QSize(18, 18))
|
||
self.expand_btn.setFixedSize(34, 34)
|
||
self.expand_btn.setCursor(QtCore.Qt.CursorShape.PointingHandCursor)
|
||
self.expand_btn.setToolTip("展开任务面板")
|
||
cl.addWidget(self.expand_btn, 0, QtCore.Qt.AlignmentFlag.AlignHCenter)
|
||
cl.addStretch(1)
|
||
self.stack.addWidget(self.collapse_page)
|
||
|
||
# 两个按钮分居 QStackedWidget 两页 → 任何时刻可见开关恒为 1 个
|
||
self.stack.setCurrentWidget(self.collapse_page)
|
||
self.fold_btn.clicked.connect(self.toggle)
|
||
self.expand_btn.clicked.connect(self.toggle)
|
||
|
||
# ---------------- 两栏折叠/展开：抽屉式高度动画（头部行高恒定）----------------
|
||
def _toggle_section(self, sec):
|
||
"""点栏头 → 只把「下面填充的 bash 层」的高度收到 0，头部行高度不变"""
|
||
self._finish_fold() # 连点：先把上一段动画落定
|
||
idx = 0 if sec is self.sec_running else 1
|
||
sizes = self.splitter.sizes()
|
||
avail = sum(sizes)
|
||
head_h = sec.head_height()
|
||
other = self.sec_done if idx == 0 else self.sec_running
|
||
if not sec.folded and not other.folded:
|
||
# 即将收起 + 另一栏展开 → 记住当前高度（用于下次展开恢复）；
|
||
# 另一栏已收起时不记（此时本栏高度是「余量」而非用户意图高度）
|
||
self._saved_h[idx] = max(sizes[idx], head_h + self.MIN_BODY_H)
|
||
sec.set_folded(not sec.folded)
|
||
target = self._layout_targets(actor=sec)
|
||
if not sec.folded: # 展开 → 内容从透明开始淡入
|
||
sec.show_body()
|
||
sec.set_body_opacity(0.0)
|
||
self._sync_handle()
|
||
if avail <= 0 or self.splitter.width() <= 0:
|
||
self.splitter.setSizes(target)
|
||
self._end_fold_visual(sec)
|
||
return
|
||
self._fold = {"sec": sec, "from": list(sizes), "to": target, "t": 0.0}
|
||
self._fold_timer.start()
|
||
|
||
def _layout_targets(self, actor=None):
|
||
"""统一布局策略（层从顶部堆叠）：
|
||
· 收起栏 = 只剩头部行
|
||
· 本次被点开的那栏 → 用记录的高度
|
||
· 余量只由「展开着的已完成」吸收（填满到底部）；
|
||
若「已完成」已收起 → 余量进底部留白（栏头紧跟上方内容）
|
||
"""
|
||
s = self.splitter.sizes()
|
||
avail = sum(s)
|
||
hr = self.sec_running.head_height()
|
||
hd = self.sec_done.head_height()
|
||
h0, h1 = s[0], s[1]
|
||
run_f, done_f = self.sec_running.folded, self.sec_done.folded
|
||
if run_f:
|
||
h0 = hr
|
||
if done_f:
|
||
h1 = hd
|
||
if (not run_f) and actor is self.sec_running:
|
||
h0 = self._saved_h[0]
|
||
if (not done_f) and actor is self.sec_done:
|
||
h1 = self._saved_h[1]
|
||
if not done_f: # 已完成展开 → 吃掉全部余量
|
||
floor0 = hr if run_f else hr + self.MIN_BODY_H
|
||
h1 = max(hd + self.MIN_BODY_H, avail - max(h0, floor0))
|
||
h0 = max(hr if run_f else hr + self.MIN_BODY_H, h0)
|
||
h1 = max(hd if done_f else hd + self.MIN_BODY_H, h1)
|
||
return [h0, h1, max(0, avail - h0 - h1)]
|
||
|
||
def _fold_tick(self):
|
||
"""抽屉式每帧：ease-out 三次方插值高度 + 内容淡入淡出（同模型选择窗口）"""
|
||
d = self._fold
|
||
if d is None:
|
||
self._fold_timer.stop()
|
||
return
|
||
d["t"] = min(1.0, d["t"] + self.FOLD_STEP / self.FOLD_MS)
|
||
e = 1.0 - (1.0 - d["t"]) ** 3
|
||
a, b = d["from"], d["to"]
|
||
self.splitter.setSizes([int(round(x + (y - x) * e)) for x, y in zip(a, b)])
|
||
sec = d["sec"]
|
||
sec.set_body_opacity((1.0 - e) if sec.folded else e)
|
||
if d["t"] >= 1.0:
|
||
self._finish_fold()
|
||
|
||
def _finish_fold(self):
|
||
self._fold_timer.stop()
|
||
d = self._fold
|
||
if d is None:
|
||
return
|
||
self._fold = None
|
||
self.splitter.setSizes(d["to"])
|
||
self._end_fold_visual(d["sec"])
|
||
|
||
def _end_fold_visual(self, sec):
|
||
if sec.folded:
|
||
sec.set_body_opacity(0.0)
|
||
sec.hide_body() # 隐藏后连离屏渲染也省了
|
||
else:
|
||
sec.show_body()
|
||
sec.set_body_opacity(1.0) # 置 1 → 自动移除 effect
|
||
|
||
def _sync_handle(self):
|
||
"""两栏都展开 → 中间分隔条可拖；任一栏收起 → 禁用（避免「刚收起又被拖大」）"""
|
||
h = self.splitter.handle(1)
|
||
if h is not None:
|
||
ok = not (self.sec_running.folded or self.sec_done.folded)
|
||
h.setEnabled(ok)
|
||
h.setCursor(QtCore.Qt.CursorShape.SplitVCursor if ok
|
||
else QtCore.Qt.CursorShape.ArrowCursor)
|
||
|
||
def _repin_fold(self):
|
||
"""窗口/面板尺寸变化后，按统一策略重新钉住（收起栏 = 只剩头部行）"""
|
||
if self._fold is not None or self.collapsed:
|
||
return
|
||
if not (self.sec_running.folded or self.sec_done.folded):
|
||
return
|
||
want = self._layout_targets()
|
||
if want != self.splitter.sizes():
|
||
self.splitter.setSizes(want)
|
||
|
||
def get_panelWidth(self):
|
||
return self._w
|
||
|
||
def set_panelWidth(self, w):
|
||
self._w = int(w)
|
||
self.setFixedWidth(int(w))
|
||
|
||
panelWidth = QtCore.pyqtProperty(int, fget=get_panelWidth, fset=set_panelWidth)
|
||
|
||
def toggle(self):
|
||
self._set_collapsed(not self._target)
|
||
|
||
def _set_collapsed(self, collapsed: bool):
|
||
self._stop_anim()
|
||
self._target = collapsed
|
||
self.stack.setCurrentWidget(
|
||
self.collapse_page if collapsed else self.expand_page)
|
||
start_w = self.width()
|
||
# 展开时若记录的宽度超出当前窗口可承受范围 → 夹一下
|
||
self._end_w = self.W_COLLAPSE if collapsed else self._clamp_w(self.W_EXPAND)
|
||
end_w = self._end_w
|
||
self._anim = QtCore.QPropertyAnimation(self, b"panelWidth")
|
||
self._anim.setDuration(260)
|
||
self._anim.setStartValue(float(start_w))
|
||
self._anim.setEndValue(float(end_w))
|
||
self._anim.setEasingCurve(QtCore.QEasingCurve.Type.InOutCubic)
|
||
self._anim.finished.connect(self._on_anim_finished)
|
||
self._anim.start()
|
||
|
||
def _on_anim_finished(self):
|
||
self.setFixedWidth(self.W_COLLAPSE if self._target else self._end_w)
|
||
self.collapsed = self._target
|
||
self._stop_anim()
|
||
|
||
# ---------------- 宽度拖拽（右边界固定，左边界跟随鼠标）----------------
|
||
def _stop_anim(self):
|
||
if self._anim is not None:
|
||
try:
|
||
self._anim.stop()
|
||
self._anim.deleteLater()
|
||
except Exception:
|
||
pass
|
||
self._anim = None
|
||
|
||
def _clamp_w(self, w) -> int:
|
||
"""夹在 [W_MIN, min(W_MAX, 主窗口 50%)]（最小宽度就在这里生效）"""
|
||
hi = self.W_MAX
|
||
par = self.parentWidget()
|
||
if par is not None and par.width() > 0:
|
||
hi = min(hi, max(self.W_MIN, int(par.width() * 0.5)))
|
||
return max(self.W_MIN, min(hi, int(w)))
|
||
|
||
def resizeEvent(self, e):
|
||
super().resizeEvent(e)
|
||
h = getattr(self, "_handle", None)
|
||
if h is not None:
|
||
h.setGeometry(0, 0, _ResizeHandle.W, self.height())
|
||
h.raise_()
|
||
self._repin_fold()
|
||
|
||
def _drag_begin(self, gx):
|
||
self._stop_anim()
|
||
# 动画中途按下 → 以当前实际宽度为准确定状态
|
||
self.collapsed = self.width() <= self.W_COLLAPSE + 8
|
||
self._target = self.collapsed
|
||
self.stack.setCurrentWidget(
|
||
self.collapse_page if self.collapsed else self.expand_page)
|
||
self._drag_right = self.mapToGlobal(QtCore.QPoint(self.width(), 0)).x()
|
||
|
||
def _drag_to_global_x(self, gx):
|
||
if self._drag_right is None:
|
||
return
|
||
w = self._clamp_w(self._drag_right - int(gx))
|
||
if self.collapsed:
|
||
if w <= self.W_COLLAPSE + 8:
|
||
return # 还在收起条内拖 → 不响应
|
||
self.collapsed = False # 从收起态直接拖开 → 立即进展开态
|
||
self._target = False
|
||
self.stack.setCurrentWidget(self.expand_page)
|
||
self._w = w
|
||
self.setFixedWidth(w)
|
||
|
||
def _commit_drag_width(self):
|
||
self._drag_right = None
|
||
if self.collapsed:
|
||
return
|
||
self.W_EXPAND = self._w
|
||
if save_panel_width(self._w):
|
||
print(f"[UI]: 任务面板宽度 -> {self._w}px（已记录，下次展开自动使用）")
|
||
|
||
# ---------------- 数据 ----------------
|
||
def set_session(self, session_id, db, active_stream=None):
|
||
"""切换会话 → 全量重建。已完成来自 DB 链；未落库的在跑任务来自 active_stream。"""
|
||
self._session_id = session_id
|
||
self._db = db
|
||
self._layers.clear()
|
||
self._order.clear()
|
||
self._running.clear()
|
||
self._done.clear()
|
||
|
||
# ① DB：本会话链上的历史 bash（标注是否还在上下文内）
|
||
try:
|
||
chain = db.get_message_chain(session_id) if db else []
|
||
except Exception:
|
||
chain = []
|
||
mark_idx = -1
|
||
for i, m in enumerate(chain):
|
||
if m.get("role") == "compaction":
|
||
mark_idx = i
|
||
for i, m in enumerate(chain):
|
||
msg_id = m.get("id") or ""
|
||
for e in _iter_bash_entries(m.get("timeline")):
|
||
cid = e.get("id") or f"dbmsg_{msg_id}_{i}"
|
||
if cid in self._layers:
|
||
continue
|
||
lay = BashLayer(cid, _cmd_of(e.get("args")))
|
||
lay.set_finished(bool(e.get("ok")), e.get("result") or "")
|
||
lay.set_in_context(i > mark_idx)
|
||
self._layers[cid] = lay
|
||
self._order.append(cid)
|
||
self._done.add(cid)
|
||
|
||
# ② 内存：正在进行的这一轮（包括已开始但未落库、以及刚结束未落库的）
|
||
for e in _iter_bash_entries((active_stream or {}).get("timeline")):
|
||
cid = e.get("id")
|
||
if not cid or cid in self._layers:
|
||
continue
|
||
lay = BashLayer(cid, _cmd_of(e.get("args")))
|
||
if e.get("ok") is None:
|
||
lay.set_running(0, 0)
|
||
if e.get("result"):
|
||
lay.append_live(str(e.get("result"))) # 尽力回放（上游上限 4000 字符）
|
||
self._running.add(cid)
|
||
else:
|
||
lay.set_finished(bool(e.get("ok")), e.get("result") or "")
|
||
self._done.add(cid)
|
||
self._layers[cid] = lay
|
||
self._order.append(cid)
|
||
|
||
self._refresh()
|
||
|
||
def _refresh(self):
|
||
run_ids = [c for c in self._order if c in self._running]
|
||
done_ids = [c for c in self._order if c in self._done]
|
||
self.sec_running.set_layers([self._layers[c] for c in run_ids])
|
||
shown = done_ids[-LAYER_LIMIT:]
|
||
self.sec_done.set_layers([self._layers[c] for c in shown], total=len(done_ids))
|
||
self.lbl_total.setText(f"共 {len(self._order)}" if self._order else "")
|
||
|
||
# ---------------- 事件 ----------------
|
||
def on_started(self, call_id, name, args):
|
||
if name != "bash" or not call_id or call_id in self._layers:
|
||
return
|
||
lay = BashLayer(call_id, _cmd_of(args))
|
||
lay.set_running(0, 0)
|
||
self._layers[call_id] = lay
|
||
self._order.append(call_id)
|
||
self._running.add(call_id)
|
||
self._refresh()
|
||
|
||
def on_output(self, call_id, text):
|
||
lay = self._layers.get(call_id)
|
||
if lay is not None and lay.ok is None:
|
||
lay.append_live(text)
|
||
|
||
def on_timed(self, call_id, elapsed, timeout):
|
||
lay = self._layers.get(call_id)
|
||
if lay is not None and lay.ok is None:
|
||
lay.set_running(elapsed, timeout)
|
||
|
||
def on_finished(self, call_id, name, ok, text):
|
||
lay = self._layers.get(call_id)
|
||
if lay is None:
|
||
if name != "bash" or not call_id:
|
||
return
|
||
lay = BashLayer(call_id, "")
|
||
self._layers[call_id] = lay
|
||
self._order.append(call_id)
|
||
self._running.discard(call_id)
|
||
self._done.add(call_id)
|
||
lay.set_finished(bool(ok), text or "")
|
||
lay.set_in_context(True)
|
||
self._refresh()
|
||
|
||
def clear_all(self):
|
||
self._layers.clear()
|
||
self._order.clear()
|
||
self._running.clear()
|
||
self._done.clear()
|
||
self._refresh()
|
||
|
||
# ---------------- 测试/调试辅助 ----------------
|
||
def layer_ids(self, which="all"):
|
||
if which == "running":
|
||
return [c for c in self._order if c in self._running]
|
||
if which == "done":
|
||
return [c for c in self._order if c in self._done]
|
||
return list(self._order)
|