"""meta_resolution sticky-continuation tests.

Covers the in-memory session-keyed cache that keeps a meta-skill match
alive across follow-up turns when the LLM failed to actually emit
``meta_invoke`` on the originating turn (e.g. length-capped on
reasoning).
"""

from __future__ import annotations

import importlib
from types import SimpleNamespace
from unittest.mock import MagicMock

import pytest

from opensquilla.skills.meta.types import MetaPlan, MetaStep

mr = importlib.import_module("opensquilla.engine.steps.meta_resolution")
meta_resolution = mr.meta_resolution


def _meta_spec(*, name: str, triggers: tuple[str, ...]):
    """Build a minimal meta skill spec that ``parse_meta_plan`` accepts."""
    plan = MetaPlan(
        name=name,
        triggers=triggers,
        priority=50,
        steps=(MetaStep(id="s1", skill="paper-section-author", kind="agent"),),
    )
    spec = SimpleNamespace(
        name=name,
        kind="meta",
        triggers=list(triggers),
        composition_raw={
            "meta_priority": plan.priority,
            "steps": [{"id": "s1", "skill": "paper-section-author", "kind": "agent"}],
        },
        metadata={"opensquilla": {"meta_priority": plan.priority}},
        body="",
    )
    return spec


def _ctx(
    *,
    message: str,
    session_id: str,
    skills: list,
    metadata: dict | None = None,
    model: str = "",
    tiers: dict | None = None,
):
    loader = MagicMock()
    loader.load_all.return_value = skills
    meta = {"skill_loader": loader}
    if metadata:
        meta.update(metadata)
    return SimpleNamespace(
        message=message,
        session_key=session_id,
        metadata=meta,
        model=model,
        system_prompt="",
        config=SimpleNamespace(
            squilla_router=SimpleNamespace(tiers=tiers or {}),
            meta_skill=SimpleNamespace(enabled=True, auto_trigger=True),
        ),
        surface_kind="cli",
    )


@pytest.fixture(autouse=True)
def _clear_sticky_cache():
    """Ensure each test sees a fresh sticky cache."""
    with mr._sticky_lock:
        mr._meta_sticky_cache.clear()
    yield
    with mr._sticky_lock:
        mr._meta_sticky_cache.clear()


@pytest.mark.asyncio
async def test_fresh_match_arms_sticky_cache():
    skills = [_meta_spec(name="meta-paper-write", triggers=("帮我写篇论文",))]
    ctx = _ctx(message="帮我写篇论文", session_id="S-A", skills=skills)

    out = await meta_resolution(ctx)
    assert "meta_match" in out.metadata
    assert out.metadata.get("meta_match_sticky") is not True

    cached = mr._sticky_get("S-A")
    assert cached is not None
    assert cached["skill"] == "meta-paper-write"
    assert cached["trigger"] == "帮我写篇论文"
    assert cached["uses"] == mr._STICKY_MAX_USES


@pytest.mark.asyncio
async def test_meta_match_upgrades_low_router_tier_to_c2_entry_model():
    skills = [_meta_spec(name="meta-short-drama", triggers=("生成一个短剧",))]
    tiers = {
        "c0": {"model": "deepseek-v4-flash"},
        "c1": {"model": "deepseek-v4-flash"},
        "c2": {"model": "deepseek-v4-pro"},
        "c3": {"model": "highest-tier-model"},
    }
    ctx = _ctx(
        message="生成一个短剧，啥都行",
        session_id="S-META-UPGRADE",
        skills=skills,
        model="deepseek-v4-flash",
        tiers=tiers,
        metadata={
            "routed_tier": "c0",
            "routed_model": "deepseek-v4-flash",
            "routing_source": "v4_phase3",
            "routing_applied": True,
        },
    )

    out = await meta_resolution(ctx)

    assert out.model == "deepseek-v4-pro"
    assert out.metadata["routed_tier"] == "c2"
    assert out.metadata["routed_model"] == "deepseek-v4-pro"
    assert out.metadata["meta_required_source"] == "meta-skill-entry"
    assert out.metadata["meta_resolution_model_upgrade"] == {
        "from_model": "deepseek-v4-flash",
        "to_model": "deepseek-v4-pro",
        "to_tier": "c2",
        "source": "meta-skill-entry",
    }


@pytest.mark.asyncio
async def test_meta_upgrade_realigns_routed_provider_to_upgraded_tier():
    skills = [_meta_spec(name="meta-short-drama", triggers=("生成一个短剧",))]
    tiers = {
        "c0": {"model": "small-model-1", "provider": "provider-a"},
        "c1": {"model": "mid-model-1", "provider": "provider-a"},
        "c2": {"model": "big-model-1", "provider": "provider-b"},
        "c3": {"model": "max-model-1", "provider": "provider-b"},
    }
    ctx = _ctx(
        message="生成一个短剧，啥都行",
        session_id="S-META-PROVIDER-REALIGN",
        skills=skills,
        model="small-model-1",
        tiers=tiers,
        metadata={
            "routed_tier": "c0",
            "routed_model": "small-model-1",
            "routed_provider": "provider-a",
            "routing_source": "v4_phase3",
            "routing_applied": True,
        },
    )

    out = await meta_resolution(ctx)

    assert out.metadata["routed_tier"] == "c2"
    assert out.metadata["routed_model"] == "big-model-1"
    assert out.metadata["routed_provider"] == "provider-b"


@pytest.mark.asyncio
async def test_meta_upgrade_clears_routed_provider_when_upgraded_tier_declares_none():
    skills = [_meta_spec(name="meta-short-drama", triggers=("生成一个短剧",))]
    tiers = {
        "c0": {"model": "small-model-1", "provider": "provider-a"},
        "c1": {"model": "mid-model-1", "provider": "provider-a"},
        "c2": {"model": "big-model-1"},
        "c3": {"model": "max-model-1"},
    }
    ctx = _ctx(
        message="生成一个短剧，啥都行",
        session_id="S-META-PROVIDER-CLEAR",
        skills=skills,
        model="small-model-1",
        tiers=tiers,
        metadata={
            "routed_tier": "c0",
            "routed_model": "small-model-1",
            "routed_provider": "provider-a",
            "routing_source": "v4_phase3",
            "routing_applied": True,
        },
    )

    out = await meta_resolution(ctx)

    assert out.metadata["routed_tier"] == "c2"
    assert "routed_provider" not in out.metadata


@pytest.mark.asyncio
async def test_meta_match_without_router_tier_does_not_force_entry_model():
    skills = [_meta_spec(name="meta-short-drama", triggers=("生成一个短剧",))]
    tiers = {"c2": {"model": "deepseek-v4-pro"}}
    ctx = _ctx(
        message="生成一个短剧，啥都行",
        session_id="S-META-NO-ROUTER",
        skills=skills,
        model="operator-selected-model",
        tiers=tiers,
    )

    out = await meta_resolution(ctx)

    assert out.model == "operator-selected-model"
    assert "meta_resolution_model_upgrade" not in out.metadata


@pytest.mark.asyncio
async def test_followup_without_trigger_replays_from_sticky():
    skills = [_meta_spec(name="meta-paper-write", triggers=("帮我写篇论文",))]
    # T1: arm cache.
    await meta_resolution(_ctx(
        message="帮我写篇论文", session_id="S-B", skills=skills,
    ))

    # T2: same session, follow-up text does not contain the trigger.
    out = await meta_resolution(_ctx(
        message="我想写一个关于 RAG 的论文，20 页左右",
        session_id="S-B",
        skills=skills,
    ))
    assert "meta_match" in out.metadata
    assert out.metadata.get("meta_match_sticky") is True
    assert out.metadata.get("meta_match").plan.name == "meta-paper-write"
    # uses decremented by 1
    assert mr._sticky_get("S-B")["uses"] == mr._STICKY_MAX_USES - 1


@pytest.mark.asyncio
async def test_sticky_replay_clamps_thinking_to_low():
    skills = [_meta_spec(name="meta-paper-write", triggers=("帮我写篇论文",))]
    ctx_fresh = _ctx(message="帮我写篇论文", session_id="S-C", skills=skills)
    out_fresh = await meta_resolution(ctx_fresh)
    assert out_fresh.metadata.get("thinking_level") == "low"
    assert out_fresh.metadata.get("thinking_source") == "meta_resolution"

    ctx_followup = _ctx(message="补充细节", session_id="S-C", skills=skills)
    out_followup = await meta_resolution(ctx_followup)
    assert out_followup.metadata.get("meta_match_sticky") is True
    assert out_followup.metadata.get("thinking_level") == "low"


@pytest.mark.asyncio
async def test_sticky_uses_decrement_to_zero_then_drops():
    skills = [_meta_spec(name="meta-paper-write", triggers=("帮我写篇论文",))]
    # Fresh match arms cache with MAX_USES.
    await meta_resolution(_ctx(
        message="帮我写篇论文", session_id="S-D", skills=skills,
    ))

    for _ in range(mr._STICKY_MAX_USES):
        out = await meta_resolution(_ctx(
            message="follow-up", session_id="S-D", skills=skills,
        ))
        assert out.metadata.get("meta_match_sticky") is True

    # After exhausting uses, the next no-trigger turn no longer matches.
    out_after = await meta_resolution(_ctx(
        message="another follow-up", session_id="S-D", skills=skills,
    ))
    assert "meta_match" not in out_after.metadata
    assert mr._sticky_get("S-D") is None


@pytest.mark.asyncio
async def test_sticky_cancel_keyword_drops_entry():
    skills = [_meta_spec(name="meta-paper-write", triggers=("帮我写篇论文",))]
    await meta_resolution(_ctx(
        message="帮我写篇论文", session_id="S-E", skills=skills,
    ))
    assert mr._sticky_get("S-E") is not None

    # User cancels — sticky cache is dropped, no replay on this turn.
    out = await meta_resolution(_ctx(
        message="算了，取消吧", session_id="S-E", skills=skills,
    ))
    assert "meta_match" not in out.metadata
    assert mr._sticky_get("S-E") is None


@pytest.mark.asyncio
async def test_pasted_context_drops_sticky_without_replay():
    skills = [_meta_spec(name="meta-paper-write", triggers=("帮我写篇论文",))]
    await meta_resolution(_ctx(
        message="帮我写篇论文", session_id="S-PASTE", skills=skills,
    ))
    assert mr._sticky_get("S-PASTE") is not None

    dump_body = "\n".join(
        [
            "WebChat dump",
            "assistant: old skill list",
            "meta-paper-write 帮我写篇论文",
        ]
        + [f"history line {i}" for i in range(12)]
    )
    out = await meta_resolution(_ctx(
        message=f"请分析下面历史页面是否误触发。\n{dump_body}",
        session_id="S-PASTE",
        skills=skills,
    ))

    assert "meta_match" not in out.metadata
    assert mr._sticky_get("S-PASTE") is None


@pytest.mark.asyncio
async def test_fresh_match_on_followup_refreshes_uses():
    """If the user re-utters the trigger, uses are re-armed."""
    skills = [_meta_spec(name="meta-paper-write", triggers=("帮我写篇论文",))]
    await meta_resolution(_ctx(
        message="帮我写篇论文", session_id="S-F", skills=skills,
    ))
    # Burn one use.
    await meta_resolution(_ctx(
        message="补充", session_id="S-F", skills=skills,
    ))
    assert mr._sticky_get("S-F")["uses"] == mr._STICKY_MAX_USES - 1

    # Re-trigger restores full budget.
    await meta_resolution(_ctx(
        message="帮我写篇论文", session_id="S-F", skills=skills,
    ))
    assert mr._sticky_get("S-F")["uses"] == mr._STICKY_MAX_USES


@pytest.mark.asyncio
async def test_sessions_are_isolated():
    skills = [_meta_spec(name="meta-paper-write", triggers=("帮我写篇论文",))]
    await meta_resolution(_ctx(
        message="帮我写篇论文", session_id="S-G1", skills=skills,
    ))
    # Different session — no sticky for it.
    out = await meta_resolution(_ctx(
        message="hello", session_id="S-G2", skills=skills,
    ))
    assert "meta_match" not in out.metadata
    assert mr._sticky_get("S-G1") is not None
    assert mr._sticky_get("S-G2") is None


@pytest.mark.asyncio
async def test_no_sticky_when_skill_was_removed():
    """If the meta-skill is no longer loaded, sticky entry is dropped."""
    skills = [_meta_spec(name="meta-paper-write", triggers=("帮我写篇论文",))]
    await meta_resolution(_ctx(
        message="帮我写篇论文", session_id="S-H", skills=skills,
    ))
    # Next turn — loader returns nothing.
    out = await meta_resolution(_ctx(
        message="follow-up", session_id="S-H", skills=[],
    ))
    assert "meta_match" not in out.metadata
    assert mr._sticky_get("S-H") is None


@pytest.mark.asyncio
async def test_skill_marketplace_intent_skips_semantic_meta_fallback(monkeypatch):
    skills = [_meta_spec(name="meta-skill-creator", triggers=("create a meta-skill",))]
    semantic_called = False

    def fake_semantic_candidate(ctx, candidates):
        nonlocal semantic_called
        semantic_called = True
        priority, name, plan, _spec = candidates[0]
        return (priority, name, plan, "semantic")

    monkeypatch.setattr(mr, "_semantic_meta_candidate", fake_semantic_candidate)

    out = await meta_resolution(_ctx(
        message="I want to install skills",
        session_id="S-I",
        skills=skills,
    ))

    assert semantic_called is False
    assert "meta_match" not in out.metadata


@pytest.mark.asyncio
async def test_skill_marketplace_intent_clears_sticky_replay():
    skills = [_meta_spec(name="meta-paper-write", triggers=("帮我写篇论文",))]
    await meta_resolution(_ctx(
        message="帮我写篇论文", session_id="S-J", skills=skills,
    ))
    assert mr._sticky_get("S-J") is not None

    out = await meta_resolution(_ctx(
        message="帮我搜索并安装 plot skill",
        session_id="S-J",
        skills=skills,
    ))

    assert "meta_match" not in out.metadata
    assert mr._sticky_get("S-J") is None


@pytest.mark.asyncio
async def test_meta_skill_discussion_clears_sticky_replay():
    skills = [_meta_spec(name="AwesomeWebpageMetaSkill", triggers=("生成图文音视频网页",))]
    await meta_resolution(_ctx(
        message="帮我生成图文音视频网页：主题是海洋塑料污染",
        session_id="S-META-DISCUSS",
        skills=skills,
    ))
    assert mr._sticky_get("S-META-DISCUSS") is not None

    out = await meta_resolution(_ctx(
        message="你最后生成的时候调用了meta skills吗，怎么生成的好差",
        session_id="S-META-DISCUSS",
        skills=skills,
    ))

    assert "meta_match" not in out.metadata
    assert mr._sticky_get("S-META-DISCUSS") is None


@pytest.mark.asyncio
async def test_meta_skill_discussion_skips_semantic_fallback(monkeypatch):
    skills = [_meta_spec(name="AwesomeWebpageMetaSkill", triggers=("生成图文音视频网页",))]
    semantic_called = False

    def fake_semantic_candidate(ctx, candidates):
        nonlocal semantic_called
        semantic_called = True
        priority, name, plan, _spec = candidates[0]
        return (priority, name, plan, "semantic")

    monkeypatch.setattr(mr, "_semantic_meta_candidate", fake_semantic_candidate)

    out = await meta_resolution(_ctx(
        message="你最后生成的时候调用了meta skills吗，怎么生成的好差",
        session_id="S-META-SEMANTIC",
        skills=skills,
    ))

    assert semantic_called is False
    assert "meta_match" not in out.metadata


@pytest.mark.asyncio
async def test_skill_marketplace_guard_does_not_block_explicit_meta_trigger():
    skills = [_meta_spec(name="meta-skill-creator", triggers=("create a meta-skill",))]

    out = await meta_resolution(_ctx(
        message="create a meta-skill that orchestrates skill search",
        session_id="S-K",
        skills=skills,
    ))

    assert out.metadata.get("meta_match").plan.name == "meta-skill-creator"