games-development-ai/cheap-worker/tests/test_ctx_compress_patch.py

"""test_ctx_compress_patch.py — 上下文压缩 fail-open 补丁单测(T5 压缩崩封口)。

守的不变量:
  · 压缩成功:透传返回值、连续失败计数清零。
  · 压缩抛异常(实证形态 jsonschema ValidationError 'important_discoveries' required):吞掉不上抛
    (生成主链继续),计连续失败。
  · 连续失败 ≥2 次:本 agent 实例断路禁用压缩(不再调原函数——压缩调用计费,防每轮白烧)。
  · apply_ctx_compress_patch:幂等(重复 apply 不套第二层);patch 后真 Agent.compress_context 带补丁标记。

跑:cheap-worker/.venv/bin/python -m pytest cheap-worker/tests/test_ctx_compress_patch.py -v
"""

import asyncio
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))  # → cheap-worker/
import _bootstrap  # noqa: E402,F401
import ctx_compress_patch as P  # noqa: E402


class _FakeAgent:
    """普通对象即可(补丁把失败状态挂实例属性,与真 Agent 同为普通类)。"""


def _run(coro):
    return asyncio.run(coro)


# ───────────────────────── 包装器行为 ─────────────────────────

def test_success_passthrough_and_reset():
    calls = []

    async def ok_orig(self, context_config=None):
        calls.append(context_config)
        return "COMPRESSED"

    wrapped = P.make_failopen_compress(ok_orig)
    a = _FakeAgent()
    assert _run(wrapped(a, context_config="CFG")) == "COMPRESSED"
    assert calls == ["CFG"]
    assert getattr(a, P._STATE_ATTR)["fails"] == 0


def test_failure_swallowed_not_raised():
    """实证崩因形态:结构化输出校验失败 → fail-open 吞掉,生成主链不被杀。"""
    async def bad_orig(self, context_config=None):
        raise ValueError("'important_discoveries' is a required property")

    wrapped = P.make_failopen_compress(bad_orig)
    a = _FakeAgent()
    assert _run(wrapped(a)) is None  # 不抛
    st = getattr(a, P._STATE_ATTR)
    assert st["fails"] == 1 and st["disabled"] is False


def test_two_consecutive_failures_open_circuit():
    """连续 2 次失败 → 断路:第 3 次起不再调原函数(压缩调用计费,断路防白烧)。"""
    call_count = {"n": 0}

    async def bad_orig(self, context_config=None):
        call_count["n"] += 1
        raise RuntimeError("compress boom")

    wrapped = P.make_failopen_compress(bad_orig)
    a = _FakeAgent()
    _run(wrapped(a))
    _run(wrapped(a))
    st = getattr(a, P._STATE_ATTR)
    assert st["disabled"] is True and st["fails"] == P.FAIL_DISABLE_THRESHOLD
    _run(wrapped(a))  # 断路后
    assert call_count["n"] == 2, "断路后不得再调原压缩(不再烧压缩调用)"


def test_success_resets_consecutive_counter():
    """失败→成功→失败:成功清零,断路器只数「连续」失败(单次抖动不致禁用)。"""
    behavior = ["fail", "ok", "fail"]

    async def flaky_orig(self, context_config=None):
        b = behavior.pop(0)
        if b == "fail":
            raise RuntimeError("boom")
        return "OK"

    wrapped = P.make_failopen_compress(flaky_orig)
    a = _FakeAgent()
    _run(wrapped(a))
    assert getattr(a, P._STATE_ATTR)["fails"] == 1
    _run(wrapped(a))
    assert getattr(a, P._STATE_ATTR)["fails"] == 0
    _run(wrapped(a))
    st = getattr(a, P._STATE_ATTR)
    assert st["fails"] == 1 and st["disabled"] is False


def test_per_agent_isolation():
    """断路状态按 agent 实例隔离:一个实例断路不连累另一个。"""
    async def bad_orig(self, context_config=None):
        raise RuntimeError("boom")

    wrapped = P.make_failopen_compress(bad_orig)
    a1, a2 = _FakeAgent(), _FakeAgent()
    _run(wrapped(a1))
    _run(wrapped(a1))
    assert getattr(a1, P._STATE_ATTR)["disabled"] is True
    _run(wrapped(a2))
    assert getattr(a2, P._STATE_ATTR)["disabled"] is False


# ───────────────────────── apply 到真 Agent(venv 2.0.2)─────────────────────────

def test_apply_patches_real_agent_and_idempotent():
    from worker import config  # noqa: F401  先走项目 import 链(代理旁路)再摸 agentscope
    from agentscope.agent import Agent

    first = P.apply_ctx_compress_patch()
    assert getattr(Agent.compress_context, P._PATCH_FLAG, False) is True, "patch 后方法应带补丁标记"
    second = P.apply_ctx_compress_patch()
    assert second is False, "重复 apply 必须幂等(不套第二层)"
    # first 可能为 False(同进程其他测试/模块已 apply 过,如 import cheap_studio),幂等语义下均合法。
    assert first in (True, False)
    assert hasattr(Agent.compress_context, "_ctx_compress_orig"), "应留原方法引用供诊断"


if __name__ == "__main__":
    _fns = [v for k, v in sorted(globals().items()) if k.startswith("test_") and callable(v)]
    _failed = 0
    for _fn in _fns:
        try:
            _fn()
            print(f"  PASS {_fn.__name__}")
        except Exception as e:  # noqa: BLE001
            _failed += 1
            print(f"  FAIL {_fn.__name__}: {type(e).__name__}: {e}")
    print(f"\n{len(_fns) - _failed}/{len(_fns)} passed")
    sys.exit(1 if _failed else 0)