projectAIpopular/tests/test_sense_promotion.py

"""晋升表测试（T-X5）：candidate→promoted→demoted→恢复状态机 + 一票否决 +
grader 晋升应用（T2→T1）与 T3/闸门互斥。"""
import asyncio

import pytest

from gateway.sense.config import build_sense_config
from gateway.sense.grader import Grader
from gateway.sense.promotion import PromotionTable, promotion_label
from gateway.sense.store import SenseStore


class _Clock:
    def __init__(self, t=1789874000.0):
        self.t = t

    def __call__(self):
        return self.t


def _table(tmp_path, **kw):
    store = SenseStore.init_db(tmp_path / "s.sqlite3")
    clock = _Clock()
    table = PromotionTable(store, now=clock, **kw)
    return store, clock, table


def test_promotion_requires_min_and_soak(tmp_path):
    """n_min 未满不晋升； soak 期未满不晋升；两者满足才 promoted。"""
    store, clock, table = _table(tmp_path, n_min=5, promote_lb=0.9,
                                 soak_days=3.0)
    label = "proxy:general"
    for _ in range(5):
        table.observe(label, ok=True, ts=clock.t)      # 通过率 1.0 但浸泡期未满
    assert not table.is_promoted(label)
    clock.t += 4 * 86400.0                              # 浸泡期满足
    table.observe(label, ok=True, ts=clock.t)
    assert table.is_promoted(label)


def test_promotion_demotes_on_degradation(tmp_path):
    """promoted 后通过率跌破 demote_lb -> 自动降级。"""
    store, clock, table = _table(tmp_path, n_min=5, promote_lb=0.9,
                                 demote_lb=0.85, soak_days=0.0)
    label = "proxy:general"
    for _ in range(6):
        table.observe(label, ok=True, ts=clock.t)
    assert table.is_promoted(label)
    for _ in range(6):
        clock.t += 60
        table.observe(label, ok=False, ts=clock.t)     # 通过率跌至 6/12 = 0.5
    assert not table.is_promoted(label)
    assert table.snapshot(label)["state"] == "demoted"


def test_promotion_recovery_after_demotion(tmp_path):
    """demoted 后凭数据恢复：再次满足晋升门 -> promoted。"""
    store, clock, table = _table(tmp_path, n_min=4, promote_lb=0.9,
                                 demote_lb=0.85, soak_days=0.0)
    label = "proxy:general"
    for _ in range(4):
        table.observe(label, ok=True, ts=clock.t)
    assert table.is_promoted(label)
    for _ in range(4):
        clock.t += 60
        table.observe(label, ok=False, ts=clock.t)     # 跌破
    assert table.snapshot(label)["state"] == "demoted"
    for _ in range(20):
        clock.t += 60
        table.observe(label, ok=True, ts=clock.t)      # 20/24 ≈ 0.83... 继续
    # 24 条中 20 ok = 0.833 < 0.9，再补 ok 至 >0.9
    for _ in range(6):
        clock.t += 60
        table.observe(label, ok=True, ts=clock.t)
    # 30 条 26 ok ≈ 0.867，仍 <0.9；再补
    for _ in range(6):
        clock.t += 60
        table.observe(label, ok=True, ts=clock.t)
    # 36 条 32 ok ≈ 0.889；最后一条后 >= 0.9
    table.observe(label, ok=True, ts=clock.t + 60)     # 37 条 33 ok ≈ 0.892
    for _ in range(4):
        table.observe(label, ok=True, ts=clock.t + 120)
    # 41 条 37 ok ≈ 0.902 >= 0.9
    assert table.snapshot(label)["state"] == "promoted"


def test_high_tier_veto(tmp_path):
    """T3 观察：不计入通过率；已晋升立即降级。"""
    store, clock, table = _table(tmp_path, n_min=3, promote_lb=0.9, soak_days=0.0)
    label = "proxy:general"
    for _ in range(3):
        table.observe(label, ok=True, ts=clock.t)
    assert table.is_promoted(label)
    before = table.snapshot(label)["n_total"]
    table.observe(label, ok=True, tier="T3", ts=clock.t + 10)
    snap = table.snapshot(label)
    assert snap["n_total"] == before                   # T3 不进统计
    assert snap["state"] == "demoted"                  # 一票否决


def test_promotion_label_format():
    assert promotion_label("proxy", "") == "proxy:general"
    assert promotion_label("proxy", "code") == "proxy:code"


# ---------------- grader 晋升应用 ----------------

class _FakeHeadMid:
    version = "test-head"

    def predict(self, vec):
        return {"t1": 0.50, "t2": 0.47, "t3": 0.03}   # 低于阈值 -> default T2


class _NoopObserver:
    def log(self, obs):
        pass


def _make(tmp_path, table):
    cfg = build_sense_config({"sense": {
        "enabled": True, "mode": "live",
        "db_path": str(tmp_path / "s.sqlite3"),
        "models_dir": str(tmp_path / "models")}})
    store = SenseStore.init_db(cfg.db_path)
    g = Grader(cfg, store, _NoopObserver(), promotion=table)
    g._load_head = lambda: _FakeHeadMid()

    async def fake_embed(text, c):
        return [1, 2, 3]

    import gateway.sense.grader as gr
    orig = gr.embed
    gr.embed = fake_embed
    return g, orig


def test_grader_applies_promotion_t2_to_t1(tmp_path):
    """已晋升标签：T2 决策升级 T1 且 hard_gates 标注 promotion_applied。"""
    store = SenseStore.init_db(tmp_path / "s.sqlite3")
    clock = _Clock()
    table = PromotionTable(store, now=clock, n_min=2, promote_lb=0.9, soak_days=0.0)
    label = promotion_label("proxy", "general")
    table.observe(label, ok=True, ts=clock.t)
    table.observe(label, ok=True, ts=clock.t)
    g, orig = _make(tmp_path, table)
    import gateway.sense.grader as gr
    try:
        d = asyncio.run(g.decide("什么是递归", "proxy"))
    finally:
        gr.embed = orig
    assert d.tier == "T1"
    assert d.hard_gates["promotion_applied"] is True


def test_grader_skips_promotion_when_not_promoted(tmp_path):
    """未晋升：同样的 T2 决策保持 T2。"""
    store = SenseStore.init_db(tmp_path / "s2.sqlite3")
    table = PromotionTable(store, now=_Clock(), n_min=100, soak_days=0.0)
    g, orig = _make(tmp_path, table)
    import gateway.sense.grader as gr
    try:
        d = asyncio.run(g.decide("什么是递归", "proxy"))
    finally:
        gr.embed = orig
    assert d.tier == "T2"
    assert d.hard_gates["promotion_applied"] is False


def test_grader_promotion_not_applied_without_table(tmp_path):
    """未接晋升表（向后兼容）：行为与既有决策一致。"""
    g, orig = _make(tmp_path, None)
    import gateway.sense.grader as gr
    try:
        d = asyncio.run(g.decide("什么是递归", "proxy"))
    finally:
        gr.embed = orig
    assert d.tier == "T2"