sanguo_moziplus_v2/tests/test_spawner_compact.py

"""单元测试：§24 v4 trajectory prompt.submitted compact 检测

测试 _check_compact_in_progress_trajectory 方法。
用 tmp_path 构造 mock trajectory jsonl 文件。
"""

import json
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pytest

from src.daemon.spawner import AgentSpawner


# ── helpers ──

_SESSION_ID = "sess-abc123"


def _make_trajectory_event(event_type: str, ts: str = None, **kwargs) -> dict:
    """构造 trajectory jsonl 事件"""
    obj = {"type": event_type}
    if ts:
        obj["ts"] = ts
    obj.update(kwargs)
    return obj


def _write_trajectory(tmp_path: Path, session_id: str, turns: list[list[dict]]):
    """写入 trajectory jsonl，按 turns 分组。

    每个 turn 是一个 list of events。
    自动在每组前加 session.started（如果该 turn 没有的话）。
    """
    traj_file = tmp_path / f"{session_id}.trajectory.jsonl"
    with open(traj_file, "w") as f:
        for turn_events in turns:
            # 如果 turn 第一个事件不是 session.started，自动加一个
            if not turn_events or turn_events[0].get("type") != "session.started":
                started = _make_trajectory_event(
                    "session.started",
                    ts=turn_events[0].get("ts") if turn_events else None,
                )
                f.write(json.dumps(started, ensure_ascii=False) + "\n")
            for evt in turn_events:
                f.write(json.dumps(evt, ensure_ascii=False) + "\n")


def _utc_now_str() -> str:
    """返回当前 UTC 时间的 ISO 字符串（带 Z 后缀）"""
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.") + \
        f"{datetime.now(timezone.utc).microsecond // 1000:03d}Z"


def _utc_past_str(minutes_ago: int) -> str:
    """返回过去 N 分钟的 UTC ISO 字符串"""
    ts = datetime.now(timezone.utc) - timedelta(minutes=minutes_ago)
    return ts.strftime("%Y-%m-%dT%H:%M:%S.") + \
        f"{ts.microsecond // 1000:03d}Z"


# ── 测试用例 ──


class TestCheckCompactInProgressTrajectory:
    """§24 v4: _check_compact_in_progress_trajectory 单元测试"""

    def test_tc1_normal_turn_with_submitted_returns_false(self, tmp_path):
        """TC1: 正常 turn（有 prompt.submitted）→ False"""
        now = _utc_now_str()
        turns = [[
            _make_trajectory_event("session.started", ts=now),
            _make_trajectory_event("context.compiled", ts=now),
            _make_trajectory_event("prompt.submitted", ts=now),
            _make_trajectory_event("model.completed", ts=now),
        ]]
        _write_trajectory(tmp_path, _SESSION_ID, turns)
        session_file = str(tmp_path / _SESSION_ID)
        assert AgentSpawner._check_compact_in_progress_trajectory(session_file) is False

    def test_tc2_compact_turn_returns_true(self, tmp_path):
        """TC2: compact turn（无 prompt.submitted，有 context.compiled + model.completed）→ True"""
        now = _utc_now_str()
        turns = [[
            _make_trajectory_event("session.started", ts=now),
            _make_trajectory_event("context.compiled", ts=now),
            _make_trajectory_event("model.completed", ts=now),
        ]]
        _write_trajectory(tmp_path, _SESSION_ID, turns)
        session_file = str(tmp_path / _SESSION_ID)
        assert AgentSpawner._check_compact_in_progress_trajectory(session_file) is True

    def test_tc3_skipped_prompt_returns_false(self, tmp_path):
        """TC3: 空白 prompt（有 prompt.skipped）→ False"""
        now = _utc_now_str()
        turns = [[
            _make_trajectory_event("session.started", ts=now),
            _make_trajectory_event("context.compiled", ts=now),
            _make_trajectory_event("prompt.skipped", ts=now),
            _make_trajectory_event("model.completed", ts=now),
        ]]
        _write_trajectory(tmp_path, _SESSION_ID, turns)
        session_file = str(tmp_path / _SESSION_ID)
        assert AgentSpawner._check_compact_in_progress_trajectory(session_file) is False

    def test_tc4_timeout_fallback_returns_false(self, tmp_path):
        """TC4: 超过 30min 兜底 → False"""
        old = _utc_past_str(35)
        turns = [[
            _make_trajectory_event("session.started", ts=old),
            _make_trajectory_event("context.compiled", ts=old),
            _make_trajectory_event("model.completed", ts=old),
        ]]
        _write_trajectory(tmp_path, _SESSION_ID, turns)
        session_file = str(tmp_path / _SESSION_ID)
        assert AgentSpawner._check_compact_in_progress_trajectory(session_file) is False

    def test_tc5_trajectory_not_exists_returns_false(self, tmp_path):
        """TC5: trajectory 不存在 → False"""
        session_file = str(tmp_path / "nonexistent-session")
        assert AgentSpawner._check_compact_in_progress_trajectory(session_file) is False

    def test_tc6_empty_trajectory_returns_false(self, tmp_path):
        """TC6: 空 trajectory → False"""
        traj_file = tmp_path / f"{_SESSION_ID}.trajectory.jsonl"
        traj_file.write_text("")
        session_file = str(tmp_path / _SESSION_ID)
        assert AgentSpawner._check_compact_in_progress_trajectory(session_file) is False

    def test_tc7_multi_turn_last_normal_returns_false(self, tmp_path):
        """TC7: 多 turn 尾部只看最后一个（最后一个正常但之前有 compact）→ False"""
        old = _utc_past_str(10)
        now = _utc_now_str()
        turn1 = [
            _make_trajectory_event("session.started", ts=old),
            _make_trajectory_event("context.compiled", ts=old),
            _make_trajectory_event("model.completed", ts=old),  # compact turn, no prompt
        ]
        turn2 = [
            _make_trajectory_event("session.started", ts=now),
            _make_trajectory_event("prompt.submitted", ts=now),
            _make_trajectory_event("model.completed", ts=now),  # normal turn
        ]
        _write_trajectory(tmp_path, _SESSION_ID, [turn1, turn2])
        session_file = str(tmp_path / _SESSION_ID)
        assert AgentSpawner._check_compact_in_progress_trajectory(session_file) is False

    def test_tc8_multi_turn_last_abnormal_returns_true(self, tmp_path):
        """TC8: 多 turn 尾部最后一个非正常 → True"""
        old = _utc_past_str(5)
        now = _utc_now_str()
        turn1 = [
            _make_trajectory_event("session.started", ts=old),
            _make_trajectory_event("prompt.submitted", ts=old),
            _make_trajectory_event("model.completed", ts=old),  # normal turn
        ]
        turn2 = [
            _make_trajectory_event("session.started", ts=now),
            _make_trajectory_event("context.compiled", ts=now),
            _make_trajectory_event("model.completed", ts=now),  # compact turn, no prompt
        ]
        _write_trajectory(tmp_path, _SESSION_ID, [turn1, turn2])
        session_file = str(tmp_path / _SESSION_ID)
        assert AgentSpawner._check_compact_in_progress_trajectory(session_file) is True

    def test_tc9_null_session_file_returns_false(self):
        """TC9: session_file 为空字符串 → False"""
        assert AgentSpawner._check_compact_in_progress_trajectory("") is False

    def test_tc10_none_session_file_returns_false(self):
        """TC10: session_file 为 None → False"""
        assert AgentSpawner._check_compact_in_progress_trajectory(None) is False