vent_agent_2.0/tests/test_core_units.py

"""核心单元测试：跟踪器、计量器、来源、提示词、工具、并发、计费、存储。

这些用例全部离线可跑（不起服务、不连模型）。端到端的 HTTP 验证见
``scripts/http_e2e.py``（需要一个真实可用的模型密钥）。
"""

from __future__ import annotations

import json

import pytest

# ============================================================
# 子智能体跟踪
# ============================================================


class TestSubagentTracker:
    def test_namespace_prefix_parsing(self):
        from api.subagent_tracker import parse_namespace

        assert parse_namespace(("tools:abc123",)) == "abc123"
        assert parse_namespace(["tools:x"]) == "x"
        assert parse_namespace("tools:y") == "y"

    def test_empty_namespace_is_main_agent(self):
        from api.subagent_tracker import parse_namespace

        assert parse_namespace(()) is None
        assert parse_namespace(None) is None
        assert parse_namespace([]) is None

    def test_control_segments_not_mistaken_for_ids(self):
        """带冒号/斜杠的控制段不能被当成 tool_call id。"""
        from api.subagent_tracker import parse_namespace

        assert parse_namespace(("model_request:1",)) is None
        assert parse_namespace(("a/b",)) is None

    def test_lifecycle_started_to_done(self):
        from api.subagent_tracker import SubagentTracker

        tracker = SubagentTracker()
        calls = [{"name": "task", "id": "c1", "args": {"subagent_type": "demo-searcher", "description": "查资料"}}]

        started = tracker.note_task_calls(calls)
        assert len(started) == 1
        assert started[0].subagent_type == "demo-searcher"
        assert started[0].status == "started"

        run = tracker.note_namespace(("tools:c1",))
        assert run is not None and run.status == "running" and run.tool_calls == 1

        done = tracker.note_result("c1", ok=True)
        assert done is not None and done.status == "done"
        assert done.duration_ms is not None

    def test_non_task_calls_ignored(self):
        from api.subagent_tracker import SubagentTracker

        tracker = SubagentTracker()
        assert tracker.note_task_calls([{"name": "calculate", "id": "c1", "args": {}}]) == []
        assert len(tracker) == 0

    def test_call_without_id_skipped(self):
        """没有 id 就无法关联后续子图事件，跳过而不是编一个假 id。"""
        from api.subagent_tracker import SubagentTracker

        tracker = SubagentTracker()
        assert tracker.note_task_calls([{"name": "task", "args": {"subagent_type": "x"}}]) == []
        assert len(tracker) == 0

    def test_unknown_namespace_does_not_invent_run(self):
        """表里没有的 id 不能凭空造一条 subagent_type=unknown 的记录。"""
        from api.subagent_tracker import SubagentTracker

        tracker = SubagentTracker()
        assert tracker.note_namespace(("tools:never-seen",)) is None
        assert len(tracker) == 0

    def test_finish_all_flushes_active(self):
        """流异常退出时不能让子智能体永远挂在 started，前端卡片会一直转。"""
        from api.subagent_tracker import SubagentTracker

        tracker = SubagentTracker()
        tracker.note_task_calls([
            {"name": "task", "id": "c1", "args": {"subagent_type": "a"}},
            {"name": "task", "id": "c2", "args": {"subagent_type": "b"}},
        ])
        tracker.note_result("c1", ok=True)
        flushed = tracker.finish_all()
        assert len(flushed) == 1 and flushed[0].call_id == "c2"
        assert not tracker.active()

    def test_duplicate_task_call_not_counted_twice(self):
        from api.subagent_tracker import SubagentTracker

        tracker = SubagentTracker()
        call = {"name": "task", "id": "c1", "args": {"subagent_type": "a"}}
        tracker.note_task_calls([call])
        tracker.note_task_calls([call])
        assert len(tracker) == 1


# ============================================================
# 上下文计量
# ============================================================


class _Msg:
    """最小的消息替身。"""

    def __init__(self, content="", usage_metadata=None, response_metadata=None,
                 additional_kwargs=None, name=None) -> None:
        self.content = content
        self.usage_metadata = usage_metadata
        self.response_metadata = response_metadata or {}
        self.additional_kwargs = additional_kwargs or {}
        self.name = name


class TestContextMeter:
    def test_extract_usage_from_langchain_metadata(self):
        from api.context_meter import extract_usage

        usage = extract_usage(_Msg(usage_metadata={
            "input_tokens": 100, "output_tokens": 20, "total_tokens": 120,
            "input_token_details": {"cache_read": 30},
        }))
        assert usage == {"input_tokens": 100, "output_tokens": 20, "cache_read_tokens": 30}

    def test_extract_usage_from_provider_raw(self):
        from api.context_meter import extract_usage

        usage = extract_usage(_Msg(response_metadata={
            "token_usage": {"prompt_tokens": 50, "completion_tokens": 5,
                            "prompt_tokens_details": {"cached_tokens": 8}}
        }))
        assert usage == {"input_tokens": 50, "output_tokens": 5, "cache_read_tokens": 8}

    def test_missing_usage_returns_none_not_zero(self):
        """「用量为 0」与「没有用量信息」必须能区分，否则估算兜底会失效。"""
        from api.context_meter import extract_usage

        assert extract_usage(_Msg()) is None
        assert extract_usage(_Msg(usage_metadata={})) is None

    def test_accumulates_across_model_calls(self):
        from api.context_meter import ContextMeter

        meter = ContextMeter(window=1000)
        meter.note_message(_Msg(usage_metadata={"input_tokens": 100, "output_tokens": 10}))
        meter.note_message(_Msg(usage_metadata={"input_tokens": 150, "output_tokens": 20}))
        assert meter.input_tokens == 250
        assert meter.output_tokens == 30
        assert meter.model_calls == 2

    def test_context_tokens_is_last_input(self):
        """上下文占用取最近一次调用的 input，而不是累加值。"""
        from api.context_meter import ContextMeter

        meter = ContextMeter(window=1000)
        meter.note_message(_Msg(usage_metadata={"input_tokens": 100, "output_tokens": 10}))
        meter.note_message(_Msg(usage_metadata={"input_tokens": 180, "output_tokens": 10}))
        assert meter.context_tokens == 180
        assert meter.input_tokens == 280

    def test_source_is_provider_when_usage_present(self):
        from api.context_meter import ContextMeter

        meter = ContextMeter(window=1000)
        meter.note_message(_Msg(usage_metadata={"input_tokens": 5, "output_tokens": 5}))
        assert meter.usage_source == "provider"

    def test_estimate_marks_source_estimate(self):
        from api.context_meter import ContextMeter

        meter = ContextMeter(window=1000)
        meter.note_estimated(300)
        assert meter.usage_source == "estimate"
        assert meter.input_tokens == 300

    def test_estimate_ignored_when_real_usage_seen(self):
        """已有真实用量时估算值只会污染口径，丢弃。"""
        from api.context_meter import ContextMeter

        meter = ContextMeter(window=1000)
        meter.note_message(_Msg(usage_metadata={"input_tokens": 100, "output_tokens": 10}))
        meter.note_estimated(9999)
        assert meter.input_tokens == 100
        assert meter.usage_source == "provider"

    def test_compaction_detected_by_lc_source(self):
        from api.context_meter import ContextMeter

        meter = ContextMeter(window=1000)
        meter.note_message(_Msg(content="摘要", response_metadata={"lc_source": "summarization"}))
        assert meter.compacted and meter.compact_count == 1

    def test_compaction_detected_by_context_drop(self):
        """占用断崖式下跌也视为压缩。"""
        from api.context_meter import ContextMeter

        meter = ContextMeter(window=10000)
        meter.note_message(_Msg(usage_metadata={"input_tokens": 9000, "output_tokens": 10}))
        assert not meter.compacted
        meter.note_message(_Msg(usage_metadata={"input_tokens": 2000, "output_tokens": 10}))
        assert meter.compacted and meter.compact_count == 1

    def test_compaction_counted_once(self):
        """两个信号同时命中不能重复计数。"""
        from api.context_meter import ContextMeter

        meter = ContextMeter(window=10000)
        meter.note_message(_Msg(usage_metadata={"input_tokens": 9000, "output_tokens": 1}))
        meter.note_message(_Msg(usage_metadata={"input_tokens": 100, "output_tokens": 1}))
        meter.note_message(_Msg(content="x", response_metadata={"lc_source": "summarization"}))
        assert meter.compact_count == 1

    def test_archive_text_captured_on_compaction(self):
        from api.context_meter import ContextMeter

        meter = ContextMeter(window=1000)
        meter.note_text("被压缩掉的早期对话")
        meter.note_message(_Msg(content="摘要", response_metadata={"lc_source": "summarization"}))
        snapshot = meter.snapshot()
        assert snapshot.compacted and snapshot.archive_text == "被压缩掉的早期对话"

    def test_fill_ratio(self):
        from api.context_meter import ContextMeter

        meter = ContextMeter(window=1000)
        meter.note_message(_Msg(usage_metadata={"input_tokens": 850, "output_tokens": 0}))
        assert meter.fill_ratio() == pytest.approx(0.85)

    def test_estimate_tokens_positive_for_cjk(self):
        from api.context_meter import estimate_tokens

        assert estimate_tokens("矿井通风") > 0
        assert estimate_tokens("") == 0


# ============================================================
# 来源标注
# ============================================================


class TestSourceTracker:
    def test_extract_from_sourced_payload(self):
        from api.source_tracker import SourceTracker
        from tools.registry import source, sourced

        tracker = SourceTracker()
        payload = json.dumps(sourced(data=[1], sources=[source("T", "https://a.invalid/x", "摘要", 0.5)]))
        added = tracker.note_tool_result("search_demo_knowledge", payload)
        assert len(added) == 1 and added[0]["title"] == "T"

    def test_extract_from_dict_payload(self):
        from api.source_tracker import SourceTracker
        from tools.registry import source, sourced

        tracker = SourceTracker()
        tracker.note_tool_result("t", sourced(data=[], sources=[source("T", "https://a.invalid")]))
        assert tracker.count == 1

    def test_dedup_by_url(self):
        from api.source_tracker import SourceTracker
        from tools.registry import source, sourced

        tracker = SourceTracker()
        payload = sourced(data=[], sources=[source("T", "https://same.invalid"), source("T2", "https://same.invalid")])
        tracker.note_tool_result("t", payload)
        assert tracker.count == 1

    def test_records_without_title_or_url_dropped(self):
        """既无标题也无链接的来源无法核实，不展示空壳。"""
        from api.source_tracker import SourceTracker
        from tools.registry import sourced

        tracker = SourceTracker()
        tracker.note_tool_result("t", sourced(data=[], sources=[{"snippet": "只有摘要"}]))
        assert tracker.count == 0

    def test_plain_text_yields_nothing(self):
        from api.source_tracker import SourceTracker

        tracker = SourceTracker()
        tracker.note_tool_result("t", "这是一段普通文本，没有 sources")
        assert tracker.count == 0

    def test_classification_web_vs_kb(self):
        from api.source_tracker import SourceTracker
        from tools.registry import source, sourced

        tracker = SourceTracker()
        tracker.note_tool_result("t", sourced(data=[], sources=[
            source("网页", "https://news.example.com/a"),
            source("知识库", "https://kb.example.com/b"),
        ]))
        types = {i["title"]: i["type"] for i in tracker.build_citations()}
        assert types["网页"] == "web"
        assert types["知识库"] == "knowledge_base"

    def test_nested_payload_lookup(self):
        from api.source_tracker import SourceTracker
        from tools.registry import source

        tracker = SourceTracker()
        tracker.note_tool_result("t", {"data": {"sources": [source("深层", "https://d.invalid")]}})
        assert tracker.count == 1 and tracker.build_citations()[0]["title"] == "深层"

    def test_citations_sorted_by_score(self):
        from api.source_tracker import SourceTracker
        from tools.registry import source, sourced

        tracker = SourceTracker()
        tracker.note_tool_result("t", sourced(data=[], sources=[
            source("低", "https://a.invalid", score=0.1),
            source("高", "https://b.invalid", score=0.9),
        ]))
        assert [i["title"] for i in tracker.build_citations()] == ["高", "低"]

    def test_label_attached(self):
        from api.source_tracker import SourceTracker
        from tools.registry import source, sourced

        tracker = SourceTracker()
        tracker.note_tool_result("search_demo_knowledge",
                                 sourced(data=[], sources=[source("T", "https://a.invalid")]))
        item = tracker.build_citations()[0]
        assert item["tool"] == "search_demo_knowledge"
        assert item["tool_cn"] == "检索演示知识库"


# ============================================================
# 提示词与消息前缀
# ============================================================


class TestPrompts:
    def test_skill_prefix_extracted(self):
        from agents.prompts import extract_skill_prefix

        text, skill = extract_skill_prefix("#skill:demo-skill 帮我写一份简报")
        assert skill == "demo-skill" and text == "帮我写一份简报"

    def test_skill_prefix_only_at_start(self):
        """正文中间的 #skill: 是普通文本，不能当指令。"""
        from agents.prompts import extract_skill_prefix

        text, skill = extract_skill_prefix("请看这个 #skill:demo-skill 的用法")
        assert skill is None and text.startswith("请看这个")

    def test_skill_prefix_without_message(self):
        from agents.prompts import extract_skill_prefix

        text, skill = extract_skill_prefix("#skill:demo-skill")
        assert skill == "demo-skill" and text == ""

    def test_path_traversal_rejected(self):
        """技能名带路径分隔符不能当技能解析——否则能读任意文件。"""
        from agents.prompts import extract_skill_prefix

        for bad in ("#skill:../../etc/passwd x", "#skill:.hidden x", "#skill:a/b x"):
            _, skill = extract_skill_prefix(bad)
            assert skill is None, bad

    def test_session_ref_extracted(self):
        from agents.prompts import extract_session_ref

        text, ref = extract_session_ref("#session:abc123 接着上次说")
        assert ref == "abc123" and text == "接着上次说"

    def test_both_prefixes_stacked(self):
        from agents.prompts import extract_session_ref, extract_skill_prefix

        text, skill = extract_skill_prefix("#skill:demo-skill #session:abc 继续")
        text, ref = extract_session_ref(text)
        assert skill == "demo-skill" and ref == "abc" and text == "继续"

    def test_mode_instruction_differs(self):
        from agents.prompts import build_mode_instruction

        plan = build_mode_instruction("plan")
        full = build_mode_instruction("full")
        assert "plan" in plan and "计划" in plan
        assert "full" in full and plan != full

    def test_preference_instruction_empty_when_no_prefs(self):
        from agents.prompts import build_preference_instruction

        assert build_preference_instruction("") == ""

    def test_time_block_format(self):
        from agents.prompts import build_time_block

        block = build_time_block()
        assert "## 系统时间" in block and "当前时间" in block

    def test_system_prompt_includes_subagents(self):
        from agents.prompts import build_system_prompt

        prompt = build_system_prompt([
            {"name": "demo-searcher", "description": "检索专员"}
        ])
        assert "demo-searcher" in prompt
        assert "检索专员" in prompt

    def test_system_prompt_without_subagents(self):
        from agents.prompts import build_system_prompt

        prompt = build_system_prompt([])
        assert "子智能体" in prompt  # 通用说明里提到，但不该有具体名单
        assert "可委派的子智能体" not in prompt


# ============================================================
# 工具
# ============================================================


class TestTools:
    def test_builtin_tools_registered(self):
        from tools import registry

        names = set(registry.get_tool_names())
        assert {"get_current_time", "search_demo_knowledge", "calculate",
                "request_plan_approval"} <= names

    def test_tools_are_basetool_instances(self):
        """子智能体的 tools 必须是 BaseTool（ToolNode 要读 .name）。"""
        from tools import registry

        for tool in registry.get_all_tools():
            assert hasattr(tool, "name") and hasattr(tool, "description")

    def test_get_tools_by_name_skips_unknown(self):
        from tools import registry

        tools = registry.get_tools_by_name(["calculate", "does-not-exist"])
        assert [t.name for t in tools] == ["calculate"]

    def test_duplicate_registration_raises(self):
        from tools.registry import register, reset_for_tests

        reset_for_tests()

        @register(group="X")
        def dup_tool() -> str:
            """d"""
            return ""

        with pytest.raises(ValueError, match="工具名冲突"):

            @register(group="Y")
            def dup_tool() -> str:  # noqa: F811
                """d"""
                return ""

        reset_for_tests()
        from tools import registry as r

        r.load_builtin_tools()

    def test_calculate_basic(self):
        from tools.demo_tools import calculate

        assert calculate("(128000 * 0.85)")["result"] == 108800

    def test_calculate_integer_result_is_int(self):
        from tools.demo_tools import calculate

        assert isinstance(calculate("1+1")["result"], int)

    def test_calculate_rejects_code_execution(self):
        """表达式求值绝不能变成代码执行入口。"""
        from tools.demo_tools import calculate

        for evil in [
            "__import__('os').system('echo x')",
            "open('/etc/passwd').read()",
            "(lambda: 1)()",
            "1 if True else 2",
            "[x for x in range(3)]",
        ]:
            result = calculate(evil)
            assert "error" in result, evil

    def test_calculate_rejects_huge_power(self):
        from tools.demo_tools import calculate

        assert "error" in calculate("9 ** 9 ** 9")

    def test_calculate_zero_division(self):
        from tools.demo_tools import calculate

        assert "零" in calculate("1/0")["error"]

    def test_calculate_empty(self):
        from tools.demo_tools import calculate

        assert "error" in calculate("")

    def test_calculate_returns_error_not_raises(self):
        """工具失败要返回可读原因，抛异常会打断整轮工具调用。"""
        from tools.demo_tools import calculate

        result = calculate("1 +")
        assert isinstance(result, dict) and "error" in result

    def test_search_returns_sources_contract(self):
        from tools.demo_tools import search_demo_knowledge

        result = search_demo_knowledge("需风量")
        assert result["data"] and result["sources"]
        assert result["sources"][0]["url"]

    def test_search_empty_query_returns_error(self):
        from tools.demo_tools import search_demo_knowledge

        result = search_demo_knowledge("   ")
        assert result["error"] and result["sources"] == []

    def test_search_no_match(self):
        from tools.demo_tools import search_demo_knowledge

        result = search_demo_knowledge("不存在的关键词zzz")
        assert result["data"] == [] and result["sources"] == []

    def test_registry_describe_shape(self):
        from tools import registry

        described = registry.describe()
        assert described["count"] >= 4
        assert all(isinstance(v, list) for v in described["groups"].values())


# ============================================================
# 并发闸
# ============================================================


class TestConcurrency:
    def setup_method(self):
        from api import concurrency

        concurrency.reset_for_tests()

    def teardown_method(self):
        from api import concurrency

        concurrency.reset_for_tests()

    def test_acquire_and_release(self):
        from api import concurrency

        slot = concurrency.acquire_task_slot("u1")
        assert concurrency.stats()["global_active"] == 1
        assert slot.release() is True
        assert concurrency.stats()["global_active"] == 0

    def test_release_is_idempotent(self):
        """SSE 有多条退出路径，重复 release 不能把计数减成负数。"""
        from api import concurrency

        slot = concurrency.acquire_task_slot("u1")
        slot.release()
        slot.release()
        assert concurrency.stats()["global_active"] == 0

    def test_per_user_limit(self, monkeypatch):
        from api import concurrency

        monkeypatch.setattr(concurrency.config, "MAX_USER_TASKS", 1)
        concurrency.acquire_task_slot("u1")
        with pytest.raises(concurrency.TooManyTasksError) as exc:
            concurrency.acquire_task_slot("u1")
        assert exc.value.scope == "user"

    def test_global_limit(self, monkeypatch):
        from api import concurrency

        monkeypatch.setattr(concurrency.config, "MAX_CONCURRENT_TASKS", 1)
        monkeypatch.setattr(concurrency.config, "MAX_USER_TASKS", 10)
        concurrency.acquire_task_slot("u1")
        with pytest.raises(concurrency.TooManyTasksError) as exc:
            concurrency.acquire_task_slot("u2")
        assert exc.value.scope == "global"

    def test_different_users_independent(self, monkeypatch):
        from api import concurrency

        monkeypatch.setattr(concurrency.config, "MAX_USER_TASKS", 1)
        concurrency.acquire_task_slot("u1")
        concurrency.acquire_task_slot("u2")
        assert concurrency.stats()["global_active"] == 2


# ============================================================
# 计费
# ============================================================


class TestBilling:
    def test_cost_computation(self, pricing_ready):
        from billing import pricing
        from db import billing_store

        billing_store.set_pricing("m1", 0.001, 0.002)
        cost, currency = pricing.compute_cost("m1", 1000, 500)
        assert cost == pytest.approx(0.001 + 0.001)
        assert currency == "CNY"

    def test_unpriced_model_fails_closed(self, pricing_ready, monkeypatch):
        """静默按 0 计费等于白送额度，所以默认必须报错。"""
        from billing import pricing

        monkeypatch.setattr(pricing.config, "get_bool", lambda *a, **k: False)
        with pytest.raises(pricing.UnpricedModelError):
            pricing.compute_cost("unknown-model", 100, 100)

    def test_unpriced_model_allowed_when_configured(self, pricing_ready, monkeypatch):
        from billing import pricing

        monkeypatch.setattr(pricing.config, "get_bool", lambda *a, **k: True)
        cost, _ = pricing.compute_cost("unknown-model", 100, 100)
        assert cost == 0.0

    def test_quota_blocks_when_exhausted(self, pricing_ready, unique_user):
        from billing import guard
        from db import billing_store, db_conn

        day = db_conn.day_key()
        billing_store.set_quota_limit(unique_user, day, 100)
        billing_store.record_usage(unique_user, "s", "chat", "m", 100, 0, 0.0, day)

        with pytest.raises(guard.QuotaExceededError):
            guard.precheck(unique_user, day)

    def test_quota_zero_means_unlimited(self, pricing_ready, unique_user):
        from billing import guard
        from db import billing_store, db_conn

        day = db_conn.day_key()
        billing_store.set_quota_limit(unique_user, day, 0)
        billing_store.record_usage(unique_user, "s", "chat", "m", 10_000_000, 0, 0.0, day)
        guard.precheck(unique_user, day)  # 不抛即通过

    def test_meter_settle_idempotent(self, pricing_ready, unique_user):
        from billing import guard
        from billing.meter import UsageMeter

        meter = UsageMeter(unique_user, "s1", "chat", "deepseek-flash")
        meter.add(1000, 100)
        first = meter.settle()
        second = meter.settle()
        assert first["settled"] is True
        assert second.get("duplicated") is True
        assert guard.snapshot(unique_user)["requests"] == 1

    def test_meter_no_usage_skips(self, pricing_ready, unique_user):
        from billing import guard
        from billing.meter import UsageMeter

        meter = UsageMeter(unique_user, "s1", "chat", "deepseek-flash")
        result = meter.settle()
        assert result["settled"] is False and result["reason"] == "no_usage"
        assert guard.snapshot(unique_user)["requests"] == 0

    def test_meter_source_downgrades_to_estimate(self, pricing_ready, unique_user):
        from billing.meter import UsageMeter

        meter = UsageMeter(unique_user, "s1", "chat", "deepseek-flash")
        meter.add(100, 10, "provider")
        meter.add(50, 5, "estimate")
        assert meter.usage_source == "estimate"

    def test_snapshot_has_total_tokens(self, pricing_ready, unique_user):
        """字段名与 billing_store.usage_for_day 对齐，消费方不用记两套名字。"""
        from billing import guard

        snapshot = guard.snapshot(unique_user)
        assert "total_tokens" in snapshot and "used_tokens" in snapshot

    def test_usage_day_bound_to_request_start(self, pricing_ready, unique_user):
        """跨零点的长请求不能被算到第二天。"""
        from billing import guard
        from billing.meter import UsageMeter
        from db import db_conn

        yesterday = "20200101"
        meter = UsageMeter(unique_user, "s1", "chat", "deepseek-flash", usage_day=yesterday)
        meter.add(100, 10)
        meter.settle()
        assert guard.snapshot(unique_user, yesterday)["requests"] == 1
        assert guard.snapshot(unique_user, db_conn.day_key())["requests"] == 0


# ============================================================
# 存储
# ============================================================


class TestStores:
    def test_message_ordering_stable_within_same_second(self):
        """同秒写入的多条消息顺序必须稳定——排序错乱会让补种的上下文角色错位。"""
        from db import chat_store

        sid = chat_store.create_session("u", "full")["id"]
        for i in range(6):
            chat_store.save_message(sid, "user", f"Q{i}")
            chat_store.save_message(sid, "assistant", f"A{i}")
        rows = chat_store.get_messages(sid)
        assert [r["msg_role"] for r in rows] == ["user", "assistant"] * 6
        assert [r["content"] for r in rows] == [
            c for i in range(6) for c in (f"Q{i}", f"A{i}")
        ]

    def test_get_messages_limit_returns_most_recent_in_order(self):
        from db import chat_store

        sid = chat_store.create_session("u", "full")["id"]
        for i in range(5):
            chat_store.save_message(sid, "user", f"M{i}")
        recent = chat_store.get_messages(sid, limit=3)
        assert [r["content"] for r in recent] == ["M2", "M3", "M4"]

    def test_message_json_fields_roundtrip(self):
        from db import chat_store

        sid = chat_store.create_session("u", "full")["id"]
        chat_store.save_message(
            sid, "assistant", "答",
            tool_calls=[{"name": "calculate", "args": {"expression": "1+1"}}],
            citations=[{"title": "T", "url": "u"}],
        )
        row = chat_store.get_messages(sid)[0]
        assert row["tool_calls"][0]["name"] == "calculate"
        assert row["citations"][0]["title"] == "T"

    def test_session_ownership_check(self):
        from db import chat_store

        sid = chat_store.create_session("alice", "full")["id"]
        assert chat_store.get_session_for_user(sid, "alice") is not None
        assert chat_store.get_session_for_user(sid, "bob") is None

    def test_delete_session_cascades(self):
        from db import chat_store

        sid = chat_store.create_session("u", "full")["id"]
        chat_store.save_message(sid, "user", "x")
        chat_store.save_context_usage(sid, 1, 10, 5)
        chat_store.delete_session(sid)
        assert chat_store.get_session(sid) is None
        assert chat_store.get_messages(sid) == []
        assert chat_store.get_context_usage(sid)["history"] == []

    def test_preference_upsert(self):
        from db import chat_store

        chat_store.save_preference("u", "单位", "米")
        chat_store.save_preference("u", "单位", "千米")
        prefs = {p["pref_key"]: p["pref_value"] for p in chat_store.list_preferences("u")}
        assert prefs == {"单位": "千米"}

    def test_preferences_as_text_empty(self):
        from db import chat_store

        assert chat_store.preferences_as_text("nobody") == ""

    def test_search_sessions_by_title(self):
        from db import chat_store

        chat_store.create_session("u", "full", "通风计算讨论")
        chat_store.create_session("u", "full", "其他话题")
        found = chat_store.search_sessions("u", "通风")
        assert len(found) == 1 and "通风" in found[0]["title"]

    def test_pinned_sessions_first(self):
        from db import chat_store

        a = chat_store.create_session("u", "full", "A")["id"]
        chat_store.create_session("u", "full", "B")
        chat_store.update_session_pin(a, True)
        assert chat_store.list_sessions("u")[0]["id"] == a

    def test_context_usage_aggregates(self):
        from db import chat_store

        sid = chat_store.create_session("u", "full")["id"]
        chat_store.save_context_usage(sid, 1, 100, 10, usage_source="provider")
        chat_store.save_context_usage(sid, 2, 200, 20, compacted=True, compact_count=1,
                                      usage_source="estimate")
        usage = chat_store.get_context_usage(sid)
        assert usage["compact_count"] == 1
        assert usage["latest"]["total_tokens"] == 220
        assert usage["latest"]["usage_source"] == "estimate"

    def test_schedule_crud_and_runs(self):
        from db import schedule_store

        schedule = schedule_store.create_schedule("u", "每日", "0 9 * * *", "汇报")
        sid = schedule["id"]
        assert schedule["enabled"] is True

        run_id = schedule_store.start_run(sid, "thread-1")
        schedule_store.finish_run(run_id, "success", output="ok")
        runs = schedule_store.list_runs(sid)
        assert len(runs) == 1 and runs[0]["status"] == "success"

        updated = schedule_store.update_schedule(sid, enabled=False, title="改名")
        assert updated["enabled"] is False and updated["title"] == "改名"

        assert schedule_store.delete_schedule(sid) is True
        assert schedule_store.get_schedule(sid) is None
        assert schedule_store.list_runs(sid) == []

    def test_schedule_update_ignores_unknown_fields(self):
        from db import schedule_store

        sid = schedule_store.create_schedule("u", "T", "0 9 * * *", "P")["id"]
        schedule_store.update_schedule(sid, title="新", 恶意字段="x")
        assert schedule_store.get_schedule(sid)["title"] == "新"

    def test_init_schema_idempotent(self):
        from db import db_conn

        db_conn.init_schema()
        db_conn.init_schema(force=True)
        assert db_conn.health()["status"] == "ok"


# ============================================================
# 路由面（把「删掉 /chat/only」这个决定变成可执行的守卫）
# ============================================================


class TestRouting:
    """断言**实际对外服务**的路由面。

    这类断言的价值不在于「测试代码」，而在于把架构决定固化下来：
    没有它，将来有人「为了兼容 1.5」把 /chat/only 加回来时，只有 code review
    能拦住；有了它，CI 会拦。

    刻意断言 ``main.app.openapi()["paths"]`` 而不是 ``api.router.routes``：
    后者在本版 FastAPI 里是 ``_IncludedRouter`` 包装（不含 ``.path``），
    而且真正要守住的是**服务出去的**那套路径。顺带还能验证路由定义合法——
    schema 生成失败会在这里报出来。
    """

    def _paths(self) -> set[str]:
        from main import app

        return set(app.openapi()["paths"])

    def test_chat_entry_exists(self):
        assert "/api/chat" in self._paths()

    def test_chat_only_is_gone(self):
        """2.0 刻意删除 /chat/only：它与 /chat 行为完全相同，唯一的 kind 标签
        无人消费，且 1.5 为此付过计费 bug 与重复验证的代价。见 AGENTS.md §六。"""
        assert "/api/chat/only" not in self._paths()

    def test_interrupt_endpoints_kept(self):
        """/chat/resume 与 /chat/interrupt 承载的是不同能力，不是同一能力的
        别名，所以保留——别把它们和 /chat/only 一起删。"""
        paths = self._paths()
        assert "/api/chat/resume" in paths
        assert "/api/chat/interrupt" in paths

    def test_detail_routes_not_lost_when_removing(self):
        """删入口时最容易连带弄丢同族路由，这里把对话族钉死。"""
        chat_paths = {p for p in self._paths() if p.startswith("/api/chat")}
        assert chat_paths == {
            "/api/chat",
            "/api/chat/resume",
            "/api/chat/interrupt",
            "/api/chat/history/{session_id}",
        }, chat_paths

    def test_chat_only_route_not_defined_in_source(self):
        """源码里不得再出现该路由的装饰器。

        只在装饰器形态上断言，不搜 ``chat_only`` 这个词——它在注释与文档里
        正当地出现（用来解释为什么没有它），搜词会把说明文字也判成违规。
        """
        import inspect

        from api import chat_routes

        src = inspect.getsource(chat_routes)
        assert '@router.post("/chat/only")' not in src
        assert '"chat_only"' not in src

    def test_kind_values_are_meaningful(self):
        """kind 只剩三个有意义取值：chat / resume / scheduled。"""
        import inspect

        from api import chat_routes
        from scheduler import runner

        chat_src = inspect.getsource(chat_routes)
        assert 'kind="chat"' in chat_src
        assert 'kind="resume"' in chat_src
        assert 'kind="scheduled"' in inspect.getsource(runner)


# ============================================================
# 模型工厂
# ============================================================


class TestModelFactory:
    def test_context_window_known(self):
        from agents import model_factory

        assert model_factory.context_window("deepseek-flash") == 128_000

    def test_context_limit_override_wins(self, monkeypatch):
        from agents import model_factory

        monkeypatch.setattr(model_factory.config, "CONTEXT_LIMIT", 5000)
        assert model_factory.context_window("deepseek-flash") == 5000

    def test_unknown_model_uses_fallback(self):
        from agents import model_factory

        assert model_factory.context_window("some-unknown-model-xyz") == \
            model_factory.FALLBACK_CONTEXT_WINDOW

    def test_thinking_levels_normalized(self):
        from agents import model_factory

        assert model_factory.normalize_thinking("HIGH") == "high"
        assert model_factory.normalize_thinking("bogus") == "off"
        assert model_factory.normalize_thinking(None) in model_factory.VALID_THINKING_LEVELS

    def test_off_sends_no_extra_body(self):
        """off 时什么都不发：不同 provider 关闭思考的字段名不一致，发错会被拒。"""
        from agents import model_factory

        assert model_factory._thinking_payload("off") is None
        assert model_factory._thinking_payload("high") is not None

    def test_build_model_declares_profile(self):
        """必须显式声明上下文窗口，否则摘要中间件首次触发时会报错。"""
        from agents import model_factory

        model = model_factory.build_model()
        assert model.profile and model.profile.get("max_input_tokens")


# ============================================================
# 中断恢复翻译
# ============================================================


class TestResumeTranslation:
    def _payload(self, **kw):
        from api.models import ResumeRequest

        base = {"session_id": "s", "status": "answered"}
        base.update(kw)
        return ResumeRequest(**base)

    def test_approve(self):
        from api.chat_routes import _to_official_resume

        assert _to_official_resume(self._payload(approved=True), 1) == \
            {"decisions": [{"type": "approve"}]}

    def test_reject(self):
        from api.chat_routes import _to_official_resume

        decisions = _to_official_resume(self._payload(approved=False), 1)["decisions"]
        assert decisions[0]["type"] == "reject" and decisions[0]["message"]

    def test_cancelled_becomes_reject(self):
        from api.chat_routes import _to_official_resume

        decisions = _to_official_resume(self._payload(status="cancelled"), 1)["decisions"]
        assert decisions[0]["type"] == "reject"

    def test_answers_become_respond(self):
        from api.chat_routes import _to_official_resume

        decisions = _to_official_resume(self._payload(answers=["甲", "乙"]), 1)["decisions"]
        assert decisions[0]["type"] == "respond" and "甲" in decisions[0]["message"]

    def test_multiple_pending_actions_get_matching_decisions(self):
        """官方要求 decisions 与 action_requests 数量严格一致。"""
        from api.chat_routes import _to_official_resume

        decisions = _to_official_resume(self._payload(approved=True, interrupt_token="reviewed-batch"), 3)["decisions"]
        assert len(decisions) == 3
        assert all(d["type"] == "approve" for d in decisions)

    def test_unknown_payload_defaults_to_safe_reject(self):
        """没有明确答复时宁可拒绝执行。"""
        from api.chat_routes import _to_official_resume

        decisions = _to_official_resume(self._payload(), 1)["decisions"]
        assert decisions[0]["type"] == "reject"