"""观察埋点测试（T-G2）：批量缓冲/刷盘/三消费方记录/升级回写。"""
import asyncio
import time

import pytest

from gateway.sense.observer import Observation, Observer, get_observer, reset_observer
from gateway.sense.store import SenseStore


def test_observer_batch_flush(tmp_path):
    """log 入队 -> 100ms 内批量落盘；多条一次刷。"""
    async def scenario():
        store = SenseStore.init_db(tmp_path / "s.sqlite3")
        reset_observer()
        obs = Observer(store, flush_interval=0.05)
        await obs.start()
        for i in range(5):
            obs.log(Observation(
                request_id=f"r{i}", consumer="pipeline",
                decided_tier="T2", executed_tier="T2",
                probs={"t1": 0.2, "t2": 0.6, "t3": 0.2},
                policy_version="v0-rule", features={"turns": 1}))
        await asyncio.sleep(0.2)
        await obs.stop()
        return store

    store = asyncio.run(scenario())
    with store._lock, store._connect() as conn:
        n = conn.execute("SELECT COUNT(*) AS c FROM tier_observations").fetchone()["c"]
    assert n == 5


def test_observer_embedding_blob(tmp_path):
    """int8 BLOB 原样持久化（D-G6）。"""
    async def scenario():
        store = SenseStore.init_db(tmp_path / "s.sqlite3")
        reset_observer()
        obs = Observer(store, flush_interval=0.05)
        await obs.start()
        obs.log(Observation(request_id="rb", consumer="proxy",
                            decided_tier="T1", executed_tier="T1",
                            embedding=bytes([1, 127, 200]), policy_version="v0"))
        await asyncio.sleep(0.2)
        await obs.stop()
        return store

    store = asyncio.run(scenario())
    with store._lock, store._connect() as conn:
        row = conn.execute(
            "SELECT embedding FROM tier_observations WHERE request_id='rb'").fetchone()
    assert bytes(row["embedding"]) == bytes([1, 127, 200])


def test_get_observer_singleton(tmp_path):
    """同循环内 get_observer 返回单例；reset 后重建。"""
    async def scenario():
        store = SenseStore.init_db(tmp_path / "s.sqlite3")
        reset_observer()
        o1 = get_observer(store)
        o2 = get_observer(store)
        return o1 is o2
    assert asyncio.run(scenario()) is True


def test_upgrade_outcome_writeback(tmp_path):
    """升级阶梯回写：outcome=escalated + executed_tier 更新。"""
    store = SenseStore.init_db(tmp_path / "s.sqlite3")
    store.insert_observation({
        "request_id": "up1", "consumer": "pipeline", "decided_tier": "T1",
        "executed_tier": "T1", "probs": "{}", "policy_version": "v0",
        "features": "{}"})
    assert store.update_outcome("up1", "escalated", executed_tier="T2") is True
    with store._lock, store._connect() as conn:
        row = conn.execute(
            "SELECT outcome, executed_tier FROM tier_observations"
            " WHERE request_id='up1'").fetchone()
    assert row["outcome"] == "escalated" and row["executed_tier"] == "T2"


def test_no_query_text_persisted(tmp_path):
    """D-G6：观察表不含 query 原文（列结构断言）。"""
    store = SenseStore.init_db(tmp_path / "s.sqlite3")
    with store._lock, store._connect() as conn:
        cols = [r["name"] for r in conn.execute(
            "PRAGMA table_info(tier_observations)").fetchall()]
    assert "query" not in cols and "query_text" not in cols
    assert "embedding" in cols and "features" in cols