cowork-local/presentation/co4e/co4e_runs.py

"""Chạy flow và bảng lịch sử lượt chạy — R08-T09.

Ba chế độ chạy: cả flow, một node, hoặc từng bước thủ công. ``_topo_order`` và
``_downstream`` là phần đồ thị — chạy node nào trước, node nào phụ thuộc node
nào.

``_on_manager_event`` là nơi mọi tín hiệu từ bộ chạy nền đổ về; nó dài vì phải
phân nhánh theo loại sự kiện, không tách nhỏ được mà không làm khó đọc hơn.
"""
from __future__ import annotations

import re
from pathlib import Path
from typing import Dict, List, Optional
from PySide6.QtCore import QSize, Qt
from PySide6.QtWidgets import QMenu, QMessageBox, QTableWidget, QTableWidgetItem
from ...core import co4e
from ...i18n import tr
from ...theme import current_palette
from .co4e_workflow_crud import _LOCKED_NODE_STATUSES


class Co4ERunsMixin:
    """Phần chạy luồng của Co4E Studio: nút Chạy, ba chế độ, và bảng Flow Status.

    Nhiều luồng chạy song song được, nên mọi thứ ở đây đều đánh khoá theo
    luồng: ``_flow_runs[wf_id]`` là run của từng luồng, ``_run_logs[run_id]``
    là nhật ký nhận sự kiện của run đó. Khung vẽ chỉ phản chiếu run của luồng
    ĐANG hiện, còn các run khác vẫn chạy nền bình thường.
    """
    def _current_mode(self) -> str:
        """Chế độ chạy đang chọn: 'auto' (mặc định), 'plan' hay 'manual'."""
        return self.mode_combo.currentData() or "auto"
    def _on_mode_changed(self, *_a) -> None:
        # switching mode resets any in-progress manual sequence
        """Đổi chế độ thì huỷ chuỗi chạy thủ công đang dở và trả nhãn nút về 'Chạy'."""
        self._manual_active = False
        self._manual_order = []
        self._manual_idx = 0
        if self._cur_run_id() is None:
            self.run_btn.setText(tr("co4e.run"))
    def _on_run_clicked(self) -> None:
        # THIS flow's run is active → interrupt it (other flows keep running).
        """Bấm nút Chạy: luồng NÀY đang chạy thì dừng nó, chưa chạy thì bắt đầu theo
        chế độ đang chọn.

        Chỉ dừng run của luồng đang mở — các luồng khác không bị đụng tới.
        """
        cur = self._cur_run_id()
        if cur is not None:
            self.manager.stop(cur)
            return
        mode = self._current_mode()
        if mode == "manual":
            self._manual_run_or_advance()
        else:
            self._start_canvas_run(plan_mode=(mode == "plan"))
    def _start_canvas_run(self, *, plan_mode: bool, only: Optional[set] = None,
                          seed: Optional[Dict[str, str]] = None) -> None:
        """Bắt đầu chạy luồng trên khung vẽ.

        ``only`` giới hạn ở một nhóm bước (chạy lại một nhánh); để trống thì chạy
        cả luồng và xoá sạch trạng thái/đầu ra cũ trước. ``seed`` là đầu ra sẵn có
        đưa vào làm ngữ cảnh, để chạy tiếp một nhánh không mất kết quả phía trên.
        """
        self._sync_wf_from_canvas()
        if not self._wf.nodes:
            self.status_message.emit(tr("co4e.no_steps"))
            return
        wf_id = self._wf.id
        if only is None:
            self.canvas.reset_statuses()
            self._outputs_for(wf_id).clear()
            self._plan_bubble = None
        self._append_chat("system", tr("co4e.run_started", name=self._wf.name))
        run_id = self.manager.start(
            self._wf, skill_map=self._skill_map(), plan_mode=plan_mode,
            only_nodes=only, seed_outputs=seed or dict(self._outputs_for(wf_id)))
        self._flow_runs[wf_id] = run_id          # track THIS flow's run (parallel-safe)
        self._run_logs[run_id] = self.chat_log   # route its events to THIS flow's log
        self.run_btn.setText(tr("co4e.interrupt"))
    def _run_single(self, node_id: str) -> None:
        """Run one step (config panel "Run this step") with upstream context."""
        if self._cur_run_id() is not None:
            return
        self._start_canvas_run(plan_mode=(self._current_mode() == "plan"),
                               only={node_id}, seed=dict(self._outputs_for(self._wf.id)))
    def _run_from(self, node_id: str) -> None:
        """Chạy lại từ một bước trở đi — tức bước đó và mọi bước phía sau nó."""
        if self._cur_run_id() is not None:
            return
        self._start_canvas_run(plan_mode=(self._current_mode() == "plan"),
                               only=self._downstream(node_id), seed=dict(self._outputs_for(self._wf.id)))
    def _downstream(self, node_id: str) -> set:
        """Tập id các bước nằm sau ``node_id`` trên đồ thị (kể cả chính nó)."""
        adj: Dict[str, List[str]] = {}
        for e in self.canvas.edges():
            adj.setdefault(e.source, []).append(e.target)
        seen, stack = set(), [node_id]
        while stack:
            cur = stack.pop()
            if cur in seen:
                continue
            seen.add(cur)
            stack.extend(adj.get(cur, []))
        return seen
    def _manual_run_or_advance(self) -> None:
        """Chế độ thủ công: lần bấm đầu khởi tạo chuỗi bước, các lần sau đi tiếp một bước."""
        if not self._manual_active:
            self._sync_wf_from_canvas()
            if not self._wf.nodes:
                self.status_message.emit(tr("co4e.no_steps"))
                return
            self.canvas.reset_statuses()
            self._outputs_for(self._wf.id).clear()
            self._plan_bubble = None
            self._manual_order = self._topo_order()
            self._manual_idx = 0
            self._manual_active = True
            self._append_chat("system", tr("co4e.manual_started", name=self._wf.name))
        self._manual_step()
    def _manual_step(self) -> None:
        """Chạy đúng một bước trong chuỗi thủ công, hoặc kết thúc nếu đã hết bước."""
        if self._manual_idx >= len(self._manual_order):
            self._manual_active = False
            self.run_btn.setText(tr("co4e.run"))
            self._append_chat("system", tr("co4e.run_done"))
            return
        nid = self._manual_order[self._manual_idx]
        label = next((n.data.label for n in self.canvas.nodes() if n.id == nid), nid)
        self._append_chat("system", tr("co4e.manual_step",
                                        i=self._manual_idx + 1, n=len(self._manual_order), label=label))
        run_id = self.manager.start(
            self._wf, skill_map=self._skill_map(),
            plan_mode=False, only_nodes={nid}, seed_outputs=dict(self._outputs_for(self._wf.id)),
            manual=True)
        self._flow_runs[self._wf.id] = run_id
        self._run_logs[run_id] = self.chat_log
        self.run_btn.setText(tr("co4e.interrupt"))
    def _topo_order(self) -> List[str]:
        """Thứ tự chạy các bước: theo lớp phụ thuộc trước, trong cùng lớp thì theo"""
        nodes = self.canvas.nodes()
        edges = self.canvas.edges()
        waves = co4e.compute_waves(nodes, edges)
        y = {n.id: n.y for n in nodes}
        return sorted((n.id for n in nodes), key=lambda nid: (waves.get(nid, 0), y.get(nid, 0)))
    def _on_manager_event(self, run_id: str, ev: dict) -> None:
        # Per-flow routing: every run's events go to ITS OWN flow log (so parallel
        # runs never mix), and the canvas mirrors ONLY the run whose flow is the
        # one currently shown. Flow Status refreshes on its own via `changed`.
        """Nhận sự kiện của một run và chuyển về đúng nơi."""
        h = self.manager.get(run_id)
        run_wf = h.wf_id if h is not None else None
        log = self._run_logs.get(run_id) or self.chat_log
        shown = getattr(self, "_wf", None) is not None and run_wf == self._wf.id
        t = ev.get("type")
        if t == "node_status":
            if shown:
                nid = ev.get("node_id")
                self.canvas.update_node_status(nid, ev.get("status"))
                # If the panel is showing THIS node right now (e.g. it was
                # idle and the user had it open when the run started), keep
                # the lock in sync instead of waiting for the next click.
                if nid == getattr(self.config, "_node_id", None):
                    self.config.set_locked(
                        self.canvas.node_status(nid) in _LOCKED_NODE_STATUSES)
        elif t == "node_output":
            if run_wf is not None:
                self._outputs_for(run_wf)[ev["node_id"]] = ev.get("output", "")
            label = ev["node_id"]
            if shown:
                label = next((n.data.label for n in self.canvas.nodes() if n.id == ev["node_id"]),
                             ev["node_id"])
            elif h is not None and h.wf is not None:
                label = next((n.data.label for n in h.wf.nodes if n.id == ev["node_id"]), ev["node_id"])
            if ev.get("output"):
                bub = self._append_chat("assistant", f"**{label}**\n\n{ev['output']}", log=log)
                # Per-message token/cost footer (↓in ↑out ▤ctx $cost), like Cowork.
                self._apply_usage(bub, run_wf, ev.get("usage"))
        elif t == "node_diff":
            self._append_diff(ev.get("title", ""), ev.get("diff", ""), log=log)
        elif t == "node_plan":
            self._append_plan(ev.get("steps") or [], log=log)
        elif t == "node_tool":
            if not ev.get("ok", True):
                # A single failed tool call isn't a step failure — the agent is told
                # to recover and continue, so show it as a neutral notice (not a red
                # "Error" that reads like the whole flow crashed).
                self._append_chat("system", "⚠ " + tr("co4e.tool_failed", name=ev.get("name", "")), log=log)
        elif t in ("run_done", "run_error"):
            # Drop THIS flow's run tracking (other flows keep running in parallel).
            if run_wf is not None and self._flow_runs.get(run_wf) == run_id:
                self._flow_runs.pop(run_wf, None)
            self._run_logs.pop(run_id, None)
            if self._manual_active and shown:
                self._manual_idx += 1
                self._manual_step()
            else:
                if shown:
                    self.run_btn.setText(tr("co4e.run"))
                self._append_chat("system", tr("co4e.run_done"), log=log)
                # Clickable link to the output folder so files are one click away.
                out = (h.out_dir if h is not None and h.out_dir else "") or str(self._flow_output_root())
                try:
                    log.add_folder_link(out, tr("co4e.open_output_link"))
                    log.scroll_to_bottom()
                except Exception:  # noqa: BLE001 - link is a nicety, never fatal
                    pass
                self._notify_run_finished(run_id)      # popup: the flow finished
                if not shown and h is not None:
                    self.status_message.emit(tr("co4e.bg_done", name=h.name, status=h.status))
    def _notify_run_finished(self, run_id: str) -> None:
        """Show a non-blocking popup when a flow finishes (done / error / stopped),
        so the user is notified even if they're on another screen."""
        h = self.manager.get(run_id)
        if h is None:
            return
        from PySide6.QtWidgets import QMessageBox

        if not hasattr(self, "_run_popups"):
            self._run_popups = []
        box = QMessageBox(self)
        box.setIcon(QMessageBox.Warning if h.status == "error" else QMessageBox.Information)
        box.setWindowTitle(tr("co4e.run_done_title"))
        box.setText(tr("co4e.run_done_popup", name=h.name,
                       status=tr("co4e.status." + h.status)))
        box.setStandardButtons(QMessageBox.Ok)
        box.setModal(False)                       # non-blocking notification
        box.setAttribute(Qt.WA_DeleteOnClose, True)
        box.finished.connect(
            lambda _r=0, b=box: self._run_popups.remove(b) if b in self._run_popups else None)
        self._run_popups.append(box)              # keep a ref so it isn't GC'd
        box.show()
    def _refresh_runs(self) -> None:
        # Rebuild the always-fresh Runs table from the manager (single source of truth).
        """Dựng lại bảng Flow Status từ dữ liệu của manager."""
        if not hasattr(self, "runs_table"):
            return
        p = current_palette()
        color = {"running": p.accent, "done": p.success, "error": p.danger,
                 "stopped": p.text_muted}
        dots = {"running": "▶", "done": "✓", "error": "✕", "stopped": "■"}
        # Most-recent run at the TOP, oldest at the bottom (manager keeps runs in
        # chronological insertion order, so reverse it for display).
        runs = list(reversed(self.manager.runs()))
        t = self.runs_table
        # Preserve the selected run across the rebuild by its id (row indices shift
        # as runs are added/deleted, so a row-index restore would jump).
        sel_item = t.item(t.currentRow(), 0) if t.currentRow() >= 0 else None
        sel_id = sel_item.data(Qt.UserRole) if sel_item is not None else None
        t.setRowCount(len(runs))
        sel_row = -1
        for r, h in enumerate(runs):
            vals = [f"{dots.get(h.status, '•')} {h.name}", tr("co4e.status." + h.status),
                    h.progress_text(), h.created_by or "-", h.created_at or "-"]
            for c, val in enumerate(vals):
                it = QTableWidgetItem(str(val))
                if c == 0:
                    it.setData(Qt.UserRole, h.id)
                if c == 1:
                    from ...ui.co4e_tab import _qcolor
                    it.setForeground(_qcolor(color.get(h.status, p.text)))
                t.setItem(r, c, it)
            if h.id == sel_id:
                sel_row = r
        if sel_row >= 0:
            t.setCurrentCell(sel_row, 0)
        # The sidebar's short run list is the same data — refresh it together.
        self._refresh_side_runs()
        # Active-run count, on the sidebar heading now that the tab strip is gone.
        n = self.manager.active_count()
        label = tr("co4e.runs_tab_n", n=n) if n else tr("co4e.runs_tab")
        if hasattr(self, "flow_bar"):
            self.flow_bar.setTabText(0, label)
        head = (self._sections.get("co4e.runs_tab") or (None,))[0]
        if head is not None:
            head.setText(("▾ " if head.isChecked() else "▸ ") + label.upper())
    def _stop_selected_run(self) -> None:
        """Dừng run đang chọn; không chọn gì thì dừng toàn bộ run của workspace này."""
        row = self.runs_table.currentRow()
        it = self.runs_table.item(row, 0) if row >= 0 else None
        if it is None:
            self.manager.stop_all()
            return
        self.manager.stop(it.data(Qt.UserRole))
    def _delete_selected_run(self) -> None:
        """Delete the selected run from the Flow Status history (a running one is
        stopped first). Removes just that single entry."""
        row = self.runs_table.currentRow()
        it = self.runs_table.item(row, 0) if row >= 0 else None
        if it is None:
            self.status_message.emit(tr("co4e.select_run"))
            return
        run_id = it.data(Qt.UserRole)
        h = self.manager.get(run_id)             # stop tracking it per-flow if we were
        if h is not None and self._flow_runs.get(h.wf_id) == run_id:
            self._flow_runs.pop(h.wf_id, None)
        self._run_logs.pop(run_id, None)
        self.manager.remove(run_id)             # emits `changed` → _refresh_runs
    def _runs_context_menu(self, pos) -> None:
        """Menu chuột phải trên bảng Flow Status: mở, đổi tên, chạy lại, dừng, xoá."""
        from PySide6.QtWidgets import QMenu
        item = self.runs_table.itemAt(pos)
        if item is None:
            return
        self.runs_table.selectRow(item.row())
        menu = QMenu(self)
        menu.addAction(tr("co4e.open_run"),
                       lambda: self._open_run_from_table(self.runs_table.item(item.row(), 0)))
        it0 = self.runs_table.item(item.row(), 0)
        rid = it0.data(Qt.UserRole) if it0 is not None else None
        menu.addAction(tr("co4e.open_output"), lambda: self._open_run_output_folder(rid))
        menu.addAction(tr("co4e.rename_run"), self._rename_selected_run)
        menu.addAction(tr("co4e.delete_run"), self._delete_selected_run)
        menu.exec(self.runs_table.viewport().mapToGlobal(pos))
    def _open_run_output_folder(self, run_id) -> None:
        """Open the workspace folder a specific run wrote its files into."""
        from ...ui.osutil import open_location
        h = self.manager.get(run_id) if run_id else None
        path = Path(h.out_dir) if (h is not None and h.out_dir) else self._flow_output_root()
        if not path.exists():
            path = self._flow_output_root()
            try:
                path.mkdir(parents=True, exist_ok=True)
            except OSError:
                pass
        open_location(str(path))
    def _rename_selected_run(self) -> None:
        """Rename the selected run in Flow Status — updates the run entry AND its
        underlying saved flow / open tab so the name stays consistent everywhere."""
        row = self.runs_table.currentRow()
        it = self.runs_table.item(row, 0) if row >= 0 else None
        if it is None:
            self.status_message.emit(tr("co4e.select_run"))
            return
        run_id = it.data(Qt.UserRole)
        h = self.manager.get(run_id)
        if h is None:
            return
        from ...ui.dialog_buttons import ask_text
        new, ok = ask_text(self, tr("co4e.rename_run"),
                           tr("co4e.rename_run_label"), text=h.name)
        new = (new or "").strip()
        if not ok or not new or new == h.name:
            return
        self.manager.rename(run_id, new)          # run entry + snapshot (→ refresh)
        # Keep the underlying saved flow + any open tab in sync.
        wf = co4e.get_workflow(h.wf_id)
        if wf is not None:
            wf.name = new
            co4e.save_workflow(wf)
            self._reload_sidebar()
        for i, f in enumerate(self._flows):
            if f.id == h.wf_id:
                f.name = new
                self.flow_bar.setTabText(i + 1, new)
                break
        if self._wf.id == h.wf_id and self.name_edit.text() != new:
            self.name_edit.setText(new)           # updates _wf.name + active tab text
    def _run_selected_in_background(self) -> None:
        """Chạy luồng đang chọn trong danh sách ở chế độ nền, không mở nó lên khung vẽ."""
        wf = self._selected_wf()
        if wf is None:
            self.status_message.emit(tr("co4e.select_flow"))
            return
        self.manager.start(wf, skill_map=self._skill_map(),
                           plan_mode=(self._current_mode() == "plan"))
        # Used to jump the sidebar back to the Workflows tab; with one column
        # there is nothing to jump to — show the run that just started instead.
        self._refresh_side_runs()
        self.status_message.emit(tr("co4e.bg_started", name=wf.name))
    def _rerun_run_item(self, item) -> None:
        """Double-click a run in the history → run that flow again (in background)."""
        h = self.manager.get(item.data(Qt.UserRole))
        if h is None:
            return
        wf = self._wf_by_id(h.wf_id)
        if wf is None:
            self.status_message.emit(tr("co4e.flow_gone"))
            return
        self.manager.start(wf, skill_map=self._skill_map(),
                           plan_mode=(self._current_mode() == "plan"))
        self.status_message.emit(tr("co4e.bg_started", name=wf.name))
    def _open_run_from_table(self, item) -> None:
        """Double-click a run row in the Runs tab → open that flow's tab and show
        its live status (opens/focuses the tab; _open_flow reflects the run)."""
        id_item = self.runs_table.item(item.row(), 0)
        if id_item is None:
            return
        h = self.manager.get(id_item.data(Qt.UserRole))
        if h is None:
            return
        # Prefer the flow the run kept a reference to (works even after its tab was
        # closed or if it was never saved); fall back to resolving by id.
        wf = getattr(h, "wf", None) or self._wf_by_id(h.wf_id)
        if wf is None:
            self.status_message.emit(tr("co4e.flow_gone"))
            return
        self._open_flow(wf)
        # reflect this run's step statuses (done/error/running) on the canvas
        for nid, st in h.node_status.items():
            self.canvas.update_node_status(nid, st)
        self.status_message.emit(tr("co4e.viewing_flow", name=wf.name))