cowork-local/tests/characterization/test_co4e_run_manager_behavior.py

"""Characterization test cho lớp CŨ ``core/co4e_run_manager.py``
(``Co4ERunManager`` + ``RunHandle``) — hook ``_on_event``/``_on_finished``/
``_on_failed`` và round-trip ``RunHandle.to_record``/``RunHandle.from_record``.

VÒNG ĐỜI: đây là giàn giáo (scaffolding), không phải công trình cuối cùng.
Mục đích DUY NHẤT là làm lưới an toàn cho đợt tách ``Co4ERunManager`` thành
``domain/workflows/run_record.py::RunRecord`` (DTO thuần) +
``application/workflows/co4e_workflow_service.py::Co4EWorkflowService``
(hành vi + lifecycle + lưu đĩa) — xem ``tests/test_co4e_workflow_service.py``,
nơi lớp MỚI được bọc lại bằng test đặc tả tử tế, và nơi có
``test_new_service_produces_same_json_record_as_old_manager`` chạy CÙNG một
chuỗi thao tác trên cả hai lớp rồi so JSON ghi ra đĩa — bằng chứng "hành vi
không lệch" chạy được. File NÀY chỉ ghi lại hành vi của lớp CŨ, KHÔNG được sửa
lớp cũ để "cho khớp" test — nếu một assert dưới đây đỏ mà code cũ trông "sai",
sửa assert, không sửa ``core/co4e_run_manager.py``. Sau khi đợt tách hoàn tất
và ``core/co4e_run_manager.py`` bị xoá/deprecate hẳn, file này hết nhiệm vụ và
nên được xoá theo (không viết lại thành spec — spec test đã có sẵn ở
``tests/test_co4e_workflow_service.py``).

CÁCH CHỐT ASSERT: mọi giá trị dưới đây được lấy bằng cách CHẠY code thật qua
``.venv/Scripts/python.exe -c "..."`` rồi dán nguyên kết quả in được vào
assert (đúng quy trình ngược yêu cầu), không suy luận lý thuyết.

AN TOÀN DỮ LIỆU (BẮT BUỘC — lý do có 2 lớp phòng thủ dưới đây):
  1. ``Co4ERunManager.__init__`` gọi ``_load_history()`` đọc
     ``~/.cowork_local/co4e/run_history.json`` THẬT (``CONFIG_DIR`` là
     ``Path.home() / ".cowork_local"``, hằng số module tính MỘT LẦN lúc
     ``cowork_local.config`` được import), rồi mọi ``changed.emit()`` (từ mọi
     hook mà test này gọi) kéo theo ``_save_history()`` ghi ĐÈ file đó. Vì
     vậy, TRƯỚC khi import bất kỳ thứ gì thuộc ``cowork_local``, module này tự
     dựng một HOME giả (``_isolate_home()``, cùng kỹ thuật
     ``tools/capture_screens.py::_isolate_home()``: đặt ``USERPROFILE``/
     ``HOME`` trỏ vào một thư mục tạm, xoá ``HOMEDRIVE``/``HOMEPATH``) rồi mới
     import ``cowork_local.config``/``cowork_local.core.co4e_run_manager`` —
     một assert ngay sau import chốt rằng ``CONFIG_DIR`` thật sự nằm trong
     sandbox đó (kiểu ``tools/check_co4e.py:47``).
  2. Phòng thủ thứ hai, độc lập với (1): mỗi test còn monkeypatch
     ``Co4ERunManager._history_path`` trỏ về một file trong ``tmp_path`` CỦA
     RIÊNG NÓ. Lý do cần thêm lớp này dù đã có (1): nếu file test này được
     chạy CÙNG bộ với các file khác đã import ``cowork_local.config`` với HOME
     thật trước đó (thứ tự collect của pytest), hằng số module ``CONFIG_DIR``/
     ``CO4E_DIR`` đã bị đóng băng theo HOME thật mất rồi — xem đúng cái bẫy
     này được ghi lại trong docstring đầu
     ``tests/characterization/test_co4e_skills_panel.py``. Vá thẳng
     ``_history_path`` (đọc lại lúc GỌI, không đọc lúc import) không phụ
     thuộc thời điểm import nên luôn đúng bất kể thứ tự collect.

KHÔNG gọi ``Co4ERunManager.start()`` (spawn ``AgentWorker``/``QThread`` thật,
gọi provider AI thật, tốn tiền, ghi file thật). Mọi test dưới đây seed thẳng
vào ``manager._runs[...]`` bằng ``RunHandle`` rồi gọi ``_on_event``/
``_on_finished``/``_on_failed`` trực tiếp — đúng route hook thật mà
``start()`` nối qua ``worker.event.connect(...)`` v.v., chỉ bỏ qua phần
spawn/chạy job.

KHÔNG CẦN QApplication: ``Co4ERunManager``/``RunHandle`` chỉ dùng
``QObject``+``Signal`` đồng luồng (kết nối rồi ``emit()`` ngay trong test,
không dựng widget, không đọc kích thước, không cần app instance).

QUIRK ĐÃ ĐÓNG ĐINH (xem case tương ứng bên dưới — ĐỪNG "dọn" các chỗ này khi
tách, chúng trông như bug nhưng là hành vi đang chạy thật hôm nay):
  * ``event = Signal(str, dict)`` khai báo kiểu cứng cho tham số thứ hai. Khi
    ``_on_event(run_id, ev)`` được gọi với ``ev=None`` và ``run_id`` KHÔNG có
    trong ``_runs`` (nên nhánh xử lý dict bị bỏ qua, đi thẳng xuống
    ``self.event.emit(run_id, ev)``), Qt/Shiboken ép kiểu ``None`` thành
    ``dict`` RỖNG (``{}``) ngay tại điểm ``emit`` — listener nhận được
    ``{}`` chứ KHÔNG PHẢI ``None``. (Shiboken có in một dòng cảnh báo
    "Cannot copy-convert ... (NoneType) to C++" ra stderr, nhưng KHÔNG ném
    lỗi.) Đây là khác biệt cố ý với lớp MỚI
    (``Co4EWorkflowService``/``domain/workflows/run_record.py``), nơi không
    còn ``Signal`` nữa nên callback nhận đúng ``None`` gốc — xem
    ``tests/test_co4e_workflow_service.py::test_on_event_none_payload_does_not_raise_and_reemits_none``.
  * Khi ``ev["type"] == "run_done"`` mà ``handle.status`` KHÔNG phải
    ``"running"`` (ví dụ đã ``"stopped"``), nhánh ``if handle.status ==
    "running":`` không đổi ``status``, nhưng ``self.changed.emit()`` vẫn được
    gọi VÔ ĐIỀU KIỆN ngay sau đó (nằm ngoài ``if``) — run "đã xong" vẫn kích
    một lần refresh + một lần ghi lịch sử xuống đĩa, dù không có gì thay đổi
    trên ``handle`` đó.
  * ``RunHandle.to_record()``/``from_record()`` đổi trường ``wf`` qua lại
    thành ĐỐI TƯỢNG ``Workflow`` thật (``workflow_to_dict``/
    ``workflow_from_dict`` từ ``core/co4e.py``) — khác hẳn ``RunRecord`` mới
    (``domain/workflows/run_record.py``), nơi ``wf`` CỐ Ý được giữ nguyên là
    dict thô vì domain không được phép import ``core.co4e.Workflow``. Đây là
    một khác biệt thiết kế có chủ đích giữa bản cũ và bản mới, không phải lỗi
    port thiếu.
  * Round-trip KHÔNG đối xứng: ``status == "running"`` đọc lại từ
    ``from_record()`` bị chốt thành ``"stopped"`` (worker của nó đã mất theo
    khi app tắt giữa run), nhưng ``to_record()`` vẫn ghi đúng "running" xuống
    đĩa tại thời điểm lưu. ``from_record({})``/``from_record(None)`` mặc định
    ``status="done"`` (không phải "running") nên KHÔNG rơi vào nhánh đổi
    thành "stopped".
  * ``total`` âm bị ``max(0, total)`` kẹp về 0 ngay lúc khởi tạo
    ``RunHandle``, không giữ nguyên giá trị âm.
"""
from __future__ import annotations

import os
import tempfile
from pathlib import Path


def _isolate_home() -> Path:
    """Trỏ USERPROFILE/HOME sang một thư mục tạm TRƯỚC khi import
    ``cowork_local`` — cùng kỹ thuật ``tools/capture_screens.py::_isolate_home()``.
    Không có dữ liệu thật nào được sao chép vào đây (khác capture_screens):
    test này không cần đọc lịch sử run thật, chỉ cần KHÔNG BAO GIỜ chạm vào nó.
    """
    sandbox = Path(tempfile.mkdtemp(prefix="co4e-run-manager-test-home-"))
    (sandbox / ".cowork_local").mkdir(parents=True, exist_ok=True)
    for var in ("USERPROFILE", "HOME"):
        os.environ[var] = str(sandbox)
    os.environ.pop("HOMEDRIVE", None)
    os.environ.pop("HOMEPATH", None)
    return sandbox


_SANDBOX_HOME = _isolate_home()

import pytest  # noqa: E402

from cowork_local.config import CONFIG_DIR  # noqa: E402
from cowork_local.core.co4e import Node, Step, Workflow  # noqa: E402
from cowork_local.core.co4e_run_manager import Co4ERunManager, RunHandle  # noqa: E402

# GIỚI HẠN ĐÃ BIẾT (không phải lỗ hổng an toàn dữ liệu — xem lớp phòng thủ #2
# ở docstring đầu file, mọi test dưới đây đều tự vá ``_history_path`` bất kể
# kết quả kiểm tra này): ``CONFIG_DIR`` là hằng số module tính MỘT LẦN lúc
# ``cowork_local.config`` được import. Khi chạy CHỈ file này (đúng lệnh VERIFY
# ở đầu task), ``_isolate_home()`` ở trên chạy trước import đầu tiên nên chốt
# đúng. Khi chạy CẢ BỘ, một file khác được pytest collect trước có thể đã
# import ``cowork_local.config`` với HOME thật rồi — ``CONFIG_DIR`` khi đó đã
# đóng băng theo giá trị thật, không cách nào isolate lại được nữa từ file này
# (cùng giới hạn được ghi trong docstring đầu
# ``tests/characterization/test_co4e_skills_panel.py``). Vì vậy test dưới đây
# CHỦ ĐỘNG bỏ qua (không fail cả file, không làm mất 30 test còn lại) khi phát
# hiện giới hạn này, thay vì assert cứng ở cấp module (từng thử — gây lỗi
# collection cho TOÀN BỘ file khi chạy chung với các file khác đã import
# ``cowork_local.config`` trước).
def test_home_isolation_pins_config_dir_into_sandbox_when_first_to_import():
    if str(_SANDBOX_HOME) not in str(CONFIG_DIR):
        pytest.skip(
            "cowork_local.config da bi mot file test khac import voi HOME "
            "that TRUOC file nay trong cung phien pytest (thu tu collect) -- "
            f"CONFIG_DIR={CONFIG_DIR!r} khong con nam trong sandbox cua file "
            "nay. Day la gioi han da biet (xem docstring dau file), KHONG "
            "phai mat an toan du lieu: moi test hook trong file nay tu va "
            "thang Co4ERunManager._history_path (doc lap voi CONFIG_DIR) nen "
            "khong test nao trong file thuc su cham vao lich su run that."
        )
    assert str(_SANDBOX_HOME) in str(CONFIG_DIR)


class _Ctx:
    """Stub ``ctx``: không hook/round-trip nào trong file này đụng tới
    ``ctx.config`` (chỉ ``start()``/``_out_dir()`` — không thuộc phạm vi bọc
    của file này — mới cần nó thật)."""


@pytest.fixture
def manager(tmp_path, monkeypatch):
    # Lớp phòng thủ thứ hai (xem docstring đầu file): mỗi test có lịch sử
    # riêng trong tmp_path của chính nó, độc lập với CONFIG_DIR.
    history_path = tmp_path / "run_history.json"
    monkeypatch.setattr(Co4ERunManager, "_history_path", lambda self: history_path)
    return Co4ERunManager(_Ctx())


def _seed(manager: Co4ERunManager, run_id: str, **kw) -> RunHandle:
    defaults = dict(wf_id="wf1", name="Flow", total=3, plan_mode=False, manual=False)
    defaults.update(kw)
    h = RunHandle(run_id, **defaults)
    manager._runs[run_id] = h
    return h


def _make_workflow(node_count: int = 1, wf_id: str = "wf-x", name: str = "Flow X") -> Workflow:
    nodes = [Node(id=f"n{i}", x=0.0, y=0.0, data=Step(label=f"Step{i}")) for i in range(1, node_count + 1)]
    return Workflow(id=wf_id, name=name, nodes=nodes, edges=[])


# ---------------------------------------------------------------------------
# RunHandle: gia tri mac dinh / kep bien
# ---------------------------------------------------------------------------

def test_run_handle_defaults_on_construction():
    h = RunHandle("run1", "wf1", "My Flow", 3, False, False)
    assert h.status == "running"
    assert h.done == 0
    assert h.progress_text() == "0/3"
    assert h.running is True


def test_run_handle_negative_total_clamped_to_zero():
    h = RunHandle("run2", "wf2", "Flow2", -5, False, False)
    assert h.total == 0


def test_run_handle_zero_total_progress_text_falls_back_to_status():
    h = RunHandle("run3", "wf3", "Flow3", 0, False, False)
    assert h.progress_text() == "running"


# ---------------------------------------------------------------------------
# to_record / from_record
# ---------------------------------------------------------------------------

def test_to_record_contains_expected_keys_and_values():
    h = RunHandle("run1", "wf1", "My Flow", 3, False, False,
                  created_by="alice", created_at="2026-08-23 10:00", project_id="p1")
    rec = h.to_record()
    assert sorted(rec.keys()) == [
        "created_at", "created_by", "done", "error", "id", "manual", "name",
        "node_status", "out_dir", "plan_mode", "project_id", "status", "total",
        "wf", "wf_id",
    ]
    assert rec["id"] == "run1"
    assert rec["status"] == "running"
    assert rec["wf"] is None


def test_round_trip_status_running_becomes_stopped():
    h = RunHandle("run1", "wf1", "My Flow", 3, False, False)
    rec = h.to_record()
    assert rec["status"] == "running"          # to_record ghi dung "running"
    back = RunHandle.from_record(rec)
    assert back.status == "stopped"            # nhung from_record chot lai thanh "stopped"


@pytest.mark.parametrize("status", ["done", "error", "stopped"])
def test_round_trip_non_running_statuses_are_preserved(status):
    h = RunHandle("run1", "wf1", "My Flow", 3, False, False)
    h.status = status
    back = RunHandle.from_record(h.to_record())
    assert back.status == status


def test_from_record_empty_dict_uses_documented_defaults():
    h = RunHandle.from_record({})
    assert h.id == ""
    assert h.status == "done"      # quirk: khong roi vao nhanh doi thanh "stopped"
    assert h.wf is None
    assert h.node_status == {}


def test_from_record_none_treated_same_as_empty_dict():
    assert RunHandle.from_record(None).id == RunHandle.from_record({}).id
    assert RunHandle.from_record(None).status == RunHandle.from_record({}).status


def test_round_trip_workflow_snapshot_becomes_real_workflow_object():
    # quirk: khac RunRecord moi (giu wf la dict tho), RunHandle CU doi wf qua
    # lai thanh doi tuong Workflow that qua workflow_to_dict/workflow_from_dict.
    h = RunHandle("run4", "wf-x", "Flow X run", 1, False, False)
    h.wf = _make_workflow()
    rec = h.to_record()
    assert isinstance(rec["wf"], dict)         # tren dia luon la dict (JSON-able)
    assert rec["wf"]["id"] == "wf-x"
    assert rec["wf"]["name"] == "Flow X"

    back = RunHandle.from_record(rec)
    assert isinstance(back.wf, Workflow)       # nhung doc lai thanh doi tuong that
    assert back.wf.id == "wf-x"
    assert back.wf.name == "Flow X"
    assert len(back.wf.nodes) == 1
    assert back.wf.nodes[0].id == "n1"


# ---------------------------------------------------------------------------
# _on_event
# ---------------------------------------------------------------------------

def test_on_event_node_status_done_increments_progress_and_emits_changed(manager):
    h = _seed(manager, "run1")
    changed = []
    manager.changed.connect(lambda: changed.append(1))
    manager._on_event("run1", {"type": "node_status", "node_id": "n1", "status": "done"})
    assert h.node_status == {"n1": "done"}
    assert h.done == 1
    assert len(changed) == 1


def test_on_event_node_status_planned_counts_as_terminal_too(manager):
    h = _seed(manager, "run1")
    manager._on_event("run1", {"type": "node_status", "node_id": "n1", "status": "planned"})
    assert h.done == 1


def test_on_event_node_status_running_is_not_terminal(manager):
    h = _seed(manager, "run1")
    manager._on_event("run1", {"type": "node_status", "node_id": "n1", "status": "running"})
    assert h.done == 0


def test_on_event_node_status_missing_keys_stores_none_key(manager):
    h = _seed(manager, "run1")
    manager._on_event("run1", {"type": "node_status"})
    assert h.node_status == {None: None}


def test_on_event_run_done_default_ok_marks_done(manager):
    h = _seed(manager, "run1")
    manager._on_event("run1", {"type": "run_done"})
    assert h.status == "done"


def test_on_event_run_done_ok_false_marks_error(manager):
    h = _seed(manager, "run1")
    manager._on_event("run1", {"type": "run_done", "ok": False})
    assert h.status == "error"


def test_on_event_run_done_when_not_running_leaves_status_but_still_emits_changed(manager):
    # quirk (xem docstring dau file): "if status == running" khong doi status,
    # nhung changed.emit() nam NGOAI if nen van chay du khong co gi doi tren
    # handle nay.
    h = _seed(manager, "run1")
    h.status = "stopped"
    changed = []
    manager.changed.connect(lambda: changed.append(1))
    manager._on_event("run1", {"type": "run_done", "ok": False})
    assert h.status == "stopped"
    assert len(changed) == 1


def test_on_event_unknown_run_id_does_not_raise_and_still_reemits_event(manager):
    received = []
    manager.event.connect(lambda rid, ev: received.append((rid, ev)))
    manager._on_event("no-such-run", {"type": "node_status", "node_id": "n1", "status": "done"})
    assert received == [("no-such-run", {"type": "node_status", "node_id": "n1", "status": "done"})]


def test_on_event_none_payload_on_unknown_run_id_is_coerced_to_empty_dict_by_qt(manager):
    # quirk (xem docstring dau file): event = Signal(str, dict) ep None thanh
    # {} ngay tai diem emit -- khac han lop MOI (khong con Signal) nhan dung
    # None goc. Day la khac biet CO CHU Y giua ban cu va ban moi, khong phai
    # bug can sua.
    received = []
    manager.event.connect(lambda rid, ev: received.append((rid, ev)))
    manager._on_event("no-such-run", None)
    assert received == [("no-such-run", {})]


def test_on_event_none_payload_on_known_run_id_does_not_mutate_handle(manager):
    h = _seed(manager, "run1")
    received = []
    manager.event.connect(lambda rid, ev: received.append((rid, ev)))
    manager._on_event("run1", None)
    assert h.status == "running"
    assert h.node_status == {}
    assert received == [("run1", {})]      # cung bi Qt ep thanh {} nhu tren


# ---------------------------------------------------------------------------
# _on_finished
# ---------------------------------------------------------------------------

def test_on_finished_while_running_settles_to_done(manager):
    h = _seed(manager, "run1")
    changed = []
    manager.changed.connect(lambda: changed.append(1))
    manager._on_finished("run1")
    assert h.status == "done"
    assert len(changed) == 1


def test_on_finished_when_already_settled_is_a_noop(manager):
    h = _seed(manager, "run1")
    h.status = "error"
    changed = []
    manager.changed.connect(lambda: changed.append(1))
    manager._on_finished("run1")
    assert h.status == "error"
    assert len(changed) == 0


def test_on_finished_unknown_run_id_is_a_total_noop(manager):
    changed = []
    manager.changed.connect(lambda: changed.append(1))
    manager._on_finished("no-such-run")
    assert manager._runs == {}
    assert changed == []


# ---------------------------------------------------------------------------
# _on_failed
# ---------------------------------------------------------------------------

def test_on_failed_marks_error_with_message_and_emits_run_error_event(manager):
    h = _seed(manager, "run1")
    events = []
    changed = []
    manager.event.connect(lambda rid, ev: events.append((rid, ev)))
    manager.changed.connect(lambda: changed.append(1))
    manager._on_failed("run1", "boom")
    assert h.status == "error"
    assert h.error == "boom"
    assert events == [("run1", {"type": "run_error", "error": "boom"})]
    assert len(changed) == 1


def test_on_failed_overrides_status_even_when_already_settled(manager):
    h = _seed(manager, "run1")
    h.status = "done"
    manager._on_failed("run1", "late failure")
    assert h.status == "error"


def test_on_failed_unknown_run_id_is_a_total_noop(manager):
    events = []
    changed = []
    manager.event.connect(lambda rid, ev: events.append((rid, ev)))
    manager.changed.connect(lambda: changed.append(1))
    manager._on_failed("no-such-run", "err")
    assert events == []
    assert changed == []


# ---------------------------------------------------------------------------
# persistence: hook -> dia THAT (trong sandbox cua rieng tung test) -> reload
# ---------------------------------------------------------------------------

def test_changed_hook_persists_to_history_file(manager, tmp_path):
    _seed(manager, "run1")
    manager._on_event("run1", {"type": "node_status", "node_id": "n1", "status": "done"})
    path = tmp_path / "run_history.json"
    assert path.exists()
    import json
    data = json.loads(path.read_text(encoding="utf-8"))
    assert len(data["runs"]) == 1
    assert data["runs"][0]["id"] == "run1"
    assert data["runs"][0]["status"] == "running"


def test_reloading_manager_after_hook_settles_running_to_stopped(tmp_path, monkeypatch):
    history_path = tmp_path / "run_history.json"
    monkeypatch.setattr(Co4ERunManager, "_history_path", lambda self: history_path)

    mgr1 = Co4ERunManager(_Ctx())
    _seed(mgr1, "run1")
    mgr1._on_event("run1", {"type": "node_status", "node_id": "n1", "status": "done"})

    mgr2 = Co4ERunManager(_Ctx())
    assert "run1" in mgr2._runs
    assert mgr2._runs["run1"].status == "stopped"
    assert mgr2._seq == 1


def test_reloaded_seq_avoids_colliding_with_history_ids(tmp_path, monkeypatch):
    history_path = tmp_path / "run_history.json"
    monkeypatch.setattr(Co4ERunManager, "_history_path", lambda self: history_path)

    mgr1 = Co4ERunManager(_Ctx())
    _seed(mgr1, "run7")
    mgr1._on_event("run7", {"type": "node_status", "node_id": "n1", "status": "done"})

    mgr2 = Co4ERunManager(_Ctx())
    assert mgr2._seq == 7
    assert mgr2._next_id() == "run8"


def test_load_history_missing_file_is_silent_noop(tmp_path, monkeypatch):
    monkeypatch.setattr(Co4ERunManager, "_history_path", lambda self: tmp_path / "does-not-exist.json")
    mgr = Co4ERunManager(_Ctx())
    assert mgr._runs == {}
    assert mgr._seq == 0