"""Multitenancy tests for the Stage 1-3 skills/notifications work.

Covers:
- ``SkillStore`` per-user visibility, global-name uniqueness, admin override
- ``_get_editable_skill_or_raise`` route helper (owner / admin / 403 / 404 paths)
- ``cascade_skill_unpublished_or_deleted`` removes skills from referencing
  agents' ``config.yaml`` and writes notifications via the sender shim
- ``NotificationStore`` CRUD + ``unread_count``
- Curator per-user isolation — user A's transitions never touch user B's skills
"""

from __future__ import annotations

import pytest
import yaml
from fastapi import HTTPException

from app.gateway.routers.skills import _get_editable_skill_or_raise, _get_owned_skill_or_raise
from deerflow.config import paths as paths_module
from deerflow.persistence.agents import MemoryAgentStore
from deerflow.persistence.notifications import MemoryNotificationStore, set_notification_store
from deerflow.persistence.skills import MemorySkillStore
from deerflow.skills import curator_log
from deerflow.skills.cascade import cascade_skill_unpublished_or_deleted


@pytest.fixture
def isolated_paths(tmp_path, monkeypatch):
    """Redirect get_paths() at a fresh temp directory for the duration of one test."""
    monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
    monkeypatch.setattr(paths_module, "_paths", None)
    yield tmp_path
    monkeypatch.setattr(paths_module, "_paths", None)


@pytest.fixture
def notification_store():
    """Install a fresh ``MemoryNotificationStore`` as the sender backend.

    Cleared on both setup and teardown so a stale store from a previous test
    can never leak in through the global ``_store`` singleton in
    ``deerflow.persistence.notifications.sender``.
    """
    store = MemoryNotificationStore()
    set_notification_store(store)
    yield store
    set_notification_store(None)


# ---------------------------------------------------------------------------
# SkillStore: per-user visibility
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_skill_store_visibility_owner_published_and_builtin() -> None:
    store = MemorySkillStore()

    # Builtin (legacy migrated, no human owner) — visible to everyone.
    await store.ensure_legacy({"name": "builtin-skill", "owner_user_id": None, "published": True})
    # User A's private skill (not published).
    await store.create({"name": "alpha-tool", "owner_user_id": "user-a", "published": False})
    # User B's published skill — visible across users.
    await store.create({"name": "beta-tool", "owner_user_id": "user-b", "published": True})
    # User B's private skill — should be invisible to A and C.
    await store.create({"name": "gamma-tool", "owner_user_id": "user-b", "published": False})

    names_for_a = {s["name"] for s in await store.list_visible("user-a")}
    assert names_for_a == {"builtin-skill", "alpha-tool", "beta-tool"}

    names_for_c = {s["name"] for s in await store.list_visible("user-c")}
    assert names_for_c == {"builtin-skill", "beta-tool"}

    # get_visible respects the same filter.
    assert (await store.get_visible("alpha-tool", "user-a")) is not None
    assert (await store.get_visible("alpha-tool", "user-c")) is None
    assert (await store.get_visible("gamma-tool", "user-a")) is None

    # builtin flag is derived from owner_user_id IS NULL.
    builtin = await store.get_any("builtin-skill")
    assert builtin is not None and builtin["builtin"] is True
    alpha = await store.get_any("alpha-tool")
    assert alpha is not None and alpha["builtin"] is False


@pytest.mark.asyncio
async def test_skill_store_create_rejects_duplicate_name() -> None:
    """Global uniqueness — even an admin can't register a name a normal user owns."""
    store = MemorySkillStore()
    await store.create({"name": "shared", "owner_user_id": "user-a"})

    with pytest.raises(ValueError, match="already exists"):
        await store.create({"name": "shared", "owner_user_id": "user-b"})


@pytest.mark.asyncio
async def test_skill_store_only_owner_can_update_or_delete() -> None:
    store = MemorySkillStore()
    await store.create({"name": "alpha-tool", "owner_user_id": "user-a", "published": False})

    # Non-owner update returns None and does not mutate.
    assert await store.update("alpha-tool", "user-b", {"published": True}) is None
    record = await store.get_any("alpha-tool")
    assert record is not None and record["published"] is False

    # Owner can update.
    updated = await store.update("alpha-tool", "user-a", {"published": True})
    assert updated is not None and updated["published"] is True

    # Non-owner delete fails.
    assert await store.delete("alpha-tool", "user-b") is False
    assert await store.get_any("alpha-tool") is not None

    # Owner delete succeeds.
    assert await store.delete("alpha-tool", "user-a") is True
    assert await store.get_any("alpha-tool") is None


@pytest.mark.asyncio
async def test_skill_store_admin_override_can_unpublish_and_delete_any_skill() -> None:
    """``update_admin`` / ``delete_admin`` bypass ownership — including others' owned skills."""
    store = MemorySkillStore()
    await store.create({"name": "alpha-tool", "owner_user_id": "user-a", "published": True})

    # update_admin can flip published on a skill the admin doesn't own.
    updated = await store.update_admin("alpha-tool", {"published": False})
    assert updated is not None and updated["published"] is False

    # delete_admin removes it.
    assert await store.delete_admin("alpha-tool") is True
    assert await store.get_any("alpha-tool") is None

    # And no-ops on a missing name.
    assert await store.update_admin("does-not-exist", {"published": False}) is None
    assert await store.delete_admin("does-not-exist") is False


@pytest.mark.asyncio
async def test_skill_store_ensure_legacy_patches_null_owner_only() -> None:
    store = MemorySkillStore()
    # First migration: no owner → claim by admin.
    first = await store.ensure_legacy({"name": "legacy-skill", "owner_user_id": "admin-id", "published": True})
    assert first["owner_user_id"] == "admin-id"

    # Second migration: someone else tries to take ownership — refused, original wins.
    second = await store.ensure_legacy({"name": "legacy-skill", "owner_user_id": "someone-else"})
    assert second["owner_user_id"] == "admin-id"


# ---------------------------------------------------------------------------
# Route helper authorization (mirrors test_agents_backend.py contract)
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_get_editable_skill_or_raise_owner_admin_and_403() -> None:
    store = MemorySkillStore()
    await store.ensure_legacy({"name": "builtin-skill", "owner_user_id": None, "published": True})
    await store.create({"name": "owned", "owner_user_id": "user-a", "published": False})
    await store.create({"name": "others-published", "owner_user_id": "user-b", "published": True})

    # Owner can edit own skill.
    record, mode = await _get_editable_skill_or_raise(store, "owned", "user-a", is_admin=False)
    assert mode == "owner" and record["name"] == "owned"

    # Admin can edit a built-in (NULL owner).
    record, mode = await _get_editable_skill_or_raise(store, "builtin-skill", "user-a", is_admin=True)
    assert mode == "admin" and record["name"] == "builtin-skill"

    # Unlike agents, the spec lets admin take down ANY skill, including
    # another user's published one.
    record, mode = await _get_editable_skill_or_raise(store, "others-published", "user-a", is_admin=True)
    assert mode == "admin" and record["owner_user_id"] == "user-b"

    # Non-admin trying to edit a built-in → 403.
    with pytest.raises(HTTPException) as exc:
        await _get_editable_skill_or_raise(store, "builtin-skill", "user-a", is_admin=False)
    assert exc.value.status_code == 403

    # Non-admin trying to edit someone else's published skill → 403.
    with pytest.raises(HTTPException) as exc:
        await _get_editable_skill_or_raise(store, "others-published", "user-a", is_admin=False)
    assert exc.value.status_code == 403

    # Missing skill → 404 regardless of admin.
    with pytest.raises(HTTPException) as exc:
        await _get_editable_skill_or_raise(store, "missing", "user-a", is_admin=True)
    assert exc.value.status_code == 404


@pytest.mark.asyncio
async def test_get_owned_skill_or_raise_distinguishes_404_from_403() -> None:
    """Visible-but-not-owned must surface as 403 so the user sees a clear reason."""
    store = MemorySkillStore()
    await store.create({"name": "published-from-b", "owner_user_id": "user-b", "published": True})

    # Owner check on a published-but-not-owned skill returns 403, not 404.
    with pytest.raises(HTTPException) as exc:
        await _get_owned_skill_or_raise(store, "published-from-b", "user-a")
    assert exc.value.status_code == 403

    # Truly invisible / nonexistent → 404.
    with pytest.raises(HTTPException) as exc:
        await _get_owned_skill_or_raise(store, "nope", "user-a")
    assert exc.value.status_code == 404


# ---------------------------------------------------------------------------
# Cascade: detaches references + writes notifications
# ---------------------------------------------------------------------------


def _write_agent_dir(base_dir, *, agent_id: str, skills: list[str]) -> None:
    agent_dir = base_dir / "agents" / agent_id
    agent_dir.mkdir(parents=True, exist_ok=True)
    (agent_dir / "config.yaml").write_text(
        yaml.safe_dump(
            {"id": agent_id, "name": agent_id, "description": "", "skills": skills},
            default_flow_style=False,
            allow_unicode=True,
        ),
        encoding="utf-8",
    )


@pytest.mark.asyncio
async def test_cascade_unpublish_strips_skill_and_notifies_other_owners(
    isolated_paths, notification_store
) -> None:
    """User A unpublishes ``alpha-tool``. User B's agent loses the skill from its
    config and gets an in-app notification. User A's own agent loses the skill
    too but no notification (self action)."""
    agent_store = MemoryAgentStore()
    # User A is the actor; she also owns one of the referencing agents.
    await agent_store.create({"id": "agent-of-a", "name": "Agent A", "user_id": "user-a"})
    await agent_store.create({"id": "agent-of-b", "name": "Agent B", "user_id": "user-b"})
    # An agent referencing a different skill — should be untouched.
    await agent_store.create({"id": "agent-untouched", "name": "Untouched", "user_id": "user-b"})

    _write_agent_dir(isolated_paths, agent_id="agent-of-a", skills=["alpha-tool", "other"])
    _write_agent_dir(isolated_paths, agent_id="agent-of-b", skills=["alpha-tool"])
    _write_agent_dir(isolated_paths, agent_id="agent-untouched", skills=["unrelated"])

    counters = await cascade_skill_unpublished_or_deleted(
        "alpha-tool",
        actor_user_id="user-a",
        operation="unpublish",
        skill_owner_user_id="user-a",
        agent_store=agent_store,
    )

    assert counters["modified"] == 2
    assert counters["notified"] == 1  # only user-b — not the actor

    # A's agent: skill stripped, no notification for the actor.
    a_cfg = yaml.safe_load((isolated_paths / "agents" / "agent-of-a" / "config.yaml").read_text("utf-8"))
    assert "alpha-tool" not in a_cfg["skills"]
    assert "other" in a_cfg["skills"]

    # B's agent: skill stripped, notification delivered to B.
    b_cfg = yaml.safe_load((isolated_paths / "agents" / "agent-of-b" / "config.yaml").read_text("utf-8"))
    assert b_cfg["skills"] == []

    user_b_inbox = await notification_store.list_for_user("user-b")
    assert len(user_b_inbox) == 1
    note = user_b_inbox[0]
    assert note["type"] == "agent_skill_removed"
    assert note["payload"]["skill_name"] == "alpha-tool"
    assert note["payload"]["agent_id"] == "agent-of-b"
    assert note["payload"]["operation"] == "unpublish"

    # Untouched agent retains its config + no notification fired for it.
    untouched_cfg = yaml.safe_load(
        (isolated_paths / "agents" / "agent-untouched" / "config.yaml").read_text("utf-8")
    )
    assert untouched_cfg["skills"] == ["unrelated"]
    assert await notification_store.unread_count("user-a") == 0


@pytest.mark.asyncio
async def test_cascade_admin_takedown_also_notifies_skill_owner(
    isolated_paths, notification_store
) -> None:
    """Admin force-takedown emits an extra ``skill_taken_down`` notification to
    the original skill owner, on top of any per-agent notifications."""
    agent_store = MemoryAgentStore()
    await agent_store.create({"id": "agent-of-b", "name": "B", "user_id": "user-b"})
    _write_agent_dir(isolated_paths, agent_id="agent-of-b", skills=["alpha-tool"])

    await cascade_skill_unpublished_or_deleted(
        "alpha-tool",
        actor_user_id="admin-id",
        operation="admin_takedown",
        skill_owner_user_id="user-a",
        agent_store=agent_store,
    )

    # B (agent owner) gets the agent-skill-removed notification.
    assert (await notification_store.unread_count("user-b")) == 1
    # A (skill owner, not the actor) gets the takedown notification.
    a_inbox = await notification_store.list_for_user("user-a")
    assert len(a_inbox) == 1
    assert a_inbox[0]["type"] == "skill_taken_down"
    assert a_inbox[0]["payload"]["skill_name"] == "alpha-tool"
    # Admin (the actor) gets nothing.
    assert (await notification_store.unread_count("admin-id")) == 0


# ---------------------------------------------------------------------------
# Notification inbox CRUD
# ---------------------------------------------------------------------------


@pytest.mark.asyncio
async def test_notification_store_crud_and_unread_count() -> None:
    store = MemoryNotificationStore()

    await store.create({"id": "n1", "user_id": "user-a", "type": "x", "title": "t1"})
    await store.create({"id": "n2", "user_id": "user-a", "type": "x", "title": "t2"})
    await store.create({"id": "n3", "user_id": "user-b", "type": "x", "title": "t3"})

    assert (await store.unread_count("user-a")) == 2
    assert (await store.unread_count("user-b")) == 1

    # list_for_user filters by recipient and respects unread_only.
    a_items = await store.list_for_user("user-a")
    assert {item["id"] for item in a_items} == {"n1", "n2"}

    # Mark one read → unread count drops; second listing reflects state.
    assert await store.mark_read("n1", "user-a") is True
    assert (await store.unread_count("user-a")) == 1
    unread = await store.list_for_user("user-a", unread_only=True)
    assert [item["id"] for item in unread] == ["n2"]

    # Cross-user mark_read is rejected.
    assert await store.mark_read("n3", "user-a") is False
    assert (await store.unread_count("user-b")) == 1

    # mark_all_read returns the count of newly-flipped rows.
    flipped = await store.mark_all_read("user-a")
    assert flipped == 1
    assert (await store.unread_count("user-a")) == 0

    # Delete behaves the same way — owner-only.
    assert await store.delete("n3", "user-a") is False
    assert await store.delete("n3", "user-b") is True
    assert (await store.list_for_user("user-b")) == []


# ---------------------------------------------------------------------------
# Curator per-user isolation
# ---------------------------------------------------------------------------


def test_curator_apply_transitions_only_touches_owned_skills(
    isolated_paths, monkeypatch
) -> None:
    """User A running curator must not mark / archive user B's skills.

    Reproduces a minimal end-of-life scenario: both users own an idle skill,
    but only A's should transition to STALE.
    """
    from datetime import UTC, datetime, timedelta

    from deerflow.skills import curator, usage
    from deerflow.skills.storage import get_or_new_skill_storage, reset_skill_storage

    # The curator's filter helper normally queries the SkillStore via
    # ``get_session_factory``; here we substitute a static mapping so the test
    # stays SQL-free. ``apply_automatic_transitions`` is synchronous so we
    # cannot wait on the async store from inside it without nested-loop pain.
    owned_by = {
        "user-a": {"a-skill"},
        "user-b": {"b-skill"},
    }
    monkeypatch.setattr(curator, "_list_user_owned_skill_names", lambda uid: owned_by.get(uid, set()))

    # Point the on-disk skills root at the temp dir so the usage helpers write
    # to ``{tmp}/skills/.usage.json`` and not into the global singleton state.
    reset_skill_storage()
    monkeypatch.setattr(
        "deerflow.skills.storage._default_skill_storage",
        get_or_new_skill_storage(skills_path=str(isolated_paths / "skills")),
        raising=False,
    )

    # Stamp both skills as agent-created and old enough to mark stale.
    old_iso = (datetime.now(UTC) - timedelta(days=90)).isoformat()
    usage._update_entry(
        "a-skill",
        {"source": usage.SOURCE_AGENT, "created_at": old_iso, "last_used_at": old_iso, "state": usage.STATE_ACTIVE},
    )
    usage._update_entry(
        "b-skill",
        {"source": usage.SOURCE_AGENT, "created_at": old_iso, "last_used_at": old_iso, "state": usage.STATE_ACTIVE},
    )

    # Run curator transitions as user-a.
    counters = curator.apply_automatic_transitions("user-a", datetime.now(UTC))

    # Only a-skill counted as checked, only it got transitioned.
    assert counters["checked"] == 1  # owned-filter narrows the entry set
    assert counters["archived"] == 1  # 90 days idle == archive_after_days

    # The on-disk state reflects the filter.
    assert usage.get_entry("a-skill")["state"] == usage.STATE_ARCHIVED
    assert usage.get_entry("b-skill")["state"] == usage.STATE_ACTIVE

    reset_skill_storage()


def test_curator_log_buckets_records_per_user() -> None:
    """Curator log records for one user must not appear in another's view —
    guards the dict-of-deques refactor in curator_log.py."""
    curator_log.clear_all()

    curator_log.start_new_run(user_id="user-a")
    curator_log.record("info", "alpha event", user_id="user-a")

    curator_log.start_new_run(user_id="user-b")
    curator_log.record("info", "beta event", user_id="user-b")

    a_records = curator_log.get_recent(user_id="user-a")
    b_records = curator_log.get_recent(user_id="user-b")

    assert [r["message"] for r in a_records] == ["run_started", "alpha event"]
    assert [r["message"] for r in b_records] == ["run_started", "beta event"]

    # An unknown user gets an empty buffer rather than a cross-user view.
    assert curator_log.get_recent(user_id="user-c") == []

    curator_log.clear_all()


# ---------------------------------------------------------------------------
# Increment #1: Curator three-condition eligibility filter
# ---------------------------------------------------------------------------


def _seed_old_agent_skill(name: str) -> None:
    """Helper: stamp an agent-created skill as 90 days idle (archive-eligible)."""
    from datetime import UTC, datetime, timedelta

    from deerflow.skills import usage

    old_iso = (datetime.now(UTC) - timedelta(days=90)).isoformat()
    usage._update_entry(
        name,
        {
            "source": usage.SOURCE_AGENT,
            "created_at": old_iso,
            "last_used_at": old_iso,
            "state": usage.STATE_ACTIVE,
        },
    )


def _isolated_skills_root(isolated_paths, monkeypatch) -> None:
    """Helper: point the on-disk skills root at the temp directory."""
    from deerflow.skills.storage import get_or_new_skill_storage, reset_skill_storage

    reset_skill_storage()
    monkeypatch.setattr(
        "deerflow.skills.storage._default_skill_storage",
        get_or_new_skill_storage(skills_path=str(isolated_paths / "skills")),
        raising=False,
    )


def test_curator_skips_skill_referenced_by_any_agent(
    isolated_paths, monkeypatch
) -> None:
    """A skill listed in any agent's config.yaml ``skills`` field must be
    exempt from automatic archive, even if it has been idle for ages."""
    from datetime import UTC, datetime

    from deerflow.skills import curator, usage

    monkeypatch.setattr(curator, "_list_user_owned_skill_names", lambda uid: {"linked-skill"})
    monkeypatch.setattr(curator, "_is_skill_published", lambda name: False)
    # Pretend one agent's config references linked-skill.
    monkeypatch.setattr(curator, "_collect_referenced_skill_names", lambda: {"linked-skill"})
    _isolated_skills_root(isolated_paths, monkeypatch)
    _seed_old_agent_skill("linked-skill")

    counters = curator.apply_automatic_transitions("user-a", datetime.now(UTC))

    # Should be excluded from the eligible set entirely.
    assert counters["checked"] == 0
    assert counters["archived"] == 0
    assert counters["skipped_referenced"] == 1
    assert usage.get_entry("linked-skill")["state"] == usage.STATE_ACTIVE


def test_curator_skips_published_skill(isolated_paths, monkeypatch) -> None:
    """A skill marked ``published`` in SkillStore must never be auto-archived
    even when nothing references it."""
    from datetime import UTC, datetime

    from deerflow.skills import curator, usage

    monkeypatch.setattr(curator, "_list_user_owned_skill_names", lambda uid: {"public-tool"})
    monkeypatch.setattr(curator, "_collect_referenced_skill_names", lambda: set())
    # curator 现在用批量接口一次性查 published 集合,而非逐技能 _is_skill_published。
    monkeypatch.setattr(curator, "_list_published_skill_names", lambda: {"public-tool"})
    _isolated_skills_root(isolated_paths, monkeypatch)
    _seed_old_agent_skill("public-tool")

    counters = curator.apply_automatic_transitions("user-a", datetime.now(UTC))

    assert counters["checked"] == 0
    assert counters["archived"] == 0
    assert counters["skipped_published"] == 1
    assert usage.get_entry("public-tool")["state"] == usage.STATE_ACTIVE


def test_curator_archives_orphan_unpublished_skill(isolated_paths, monkeypatch) -> None:
    """The "default happy path": owner's own agent-created skill, unpublished,
    not referenced anywhere — IS archived when idle past the threshold."""
    from datetime import UTC, datetime

    from deerflow.skills import curator, usage

    monkeypatch.setattr(curator, "_list_user_owned_skill_names", lambda uid: {"orphan-skill"})
    monkeypatch.setattr(curator, "_collect_referenced_skill_names", lambda: set())
    monkeypatch.setattr(curator, "_is_skill_published", lambda name: False)
    _isolated_skills_root(isolated_paths, monkeypatch)
    _seed_old_agent_skill("orphan-skill")

    counters = curator.apply_automatic_transitions("user-a", datetime.now(UTC))

    assert counters["checked"] == 1
    assert counters["archived"] == 1
    assert counters["skipped_referenced"] == 0
    assert counters["skipped_published"] == 0
    assert usage.get_entry("orphan-skill")["state"] == usage.STATE_ARCHIVED


def test_skill_archive_tool_refuses_referenced_or_published(
    isolated_paths, monkeypatch
) -> None:
    """Safety net: even if the LLM bypasses the candidate filter and invokes
    skill_archive with a protected name, the tool itself refuses."""
    from deerflow.skills import curator

    # Make the on-disk lookup believe the skill exists and is not pinned.
    class _FakeStorage:
        def custom_skill_exists(self, _name: str) -> bool:
            return True

    class _FakeEntry:
        def get(self, key, default=None):
            return False  # pinned=False

    monkeypatch.setattr("deerflow.skills.storage.get_or_new_skill_storage", lambda *_a, **_kw: _FakeStorage())
    monkeypatch.setattr("deerflow.skills.usage.get_entry", lambda _n: _FakeEntry())

    tool = curator._make_skill_archive_tool("user-a")

    # Case A: target is referenced — must refuse.
    monkeypatch.setattr(curator, "_collect_referenced_skill_names", lambda: {"linked"})
    monkeypatch.setattr(curator, "_is_skill_published", lambda name: False)
    out_a = tool.invoke({"name": "linked", "reason": "test"})
    assert "Refused" in out_a
    assert "referenced" in out_a

    # Case B: target is published — must refuse.
    monkeypatch.setattr(curator, "_collect_referenced_skill_names", lambda: set())
    monkeypatch.setattr(curator, "_is_skill_published", lambda name: name == "pub")
    out_b = tool.invoke({"name": "pub", "reason": "test"})
    assert "Refused" in out_b
    assert "published" in out_b


def test_collect_referenced_skill_names_aggregates_from_disk(
    isolated_paths, monkeypatch
) -> None:
    """End-to-end of the scanner helper: it walks ``.deer-flow/agents/*/config.yaml``
    and returns the union of every ``skills`` list."""
    from deerflow.skills import curator

    # Build two agents on disk, each referencing one skill.
    agents_root = isolated_paths / "agents"
    (agents_root / "agent-1").mkdir(parents=True, exist_ok=True)
    (agents_root / "agent-2").mkdir(parents=True, exist_ok=True)
    (agents_root / "agent-1" / "config.yaml").write_text(
        yaml.safe_dump({"id": "agent-1", "name": "A", "skills": ["alpha", "beta"]}),
        encoding="utf-8",
    )
    (agents_root / "agent-2" / "config.yaml").write_text(
        yaml.safe_dump({"id": "agent-2", "name": "B", "skills": ["beta", "gamma"]}),
        encoding="utf-8",
    )

    # Point get_paths().agents_dir at the temp tree.
    class _FakePaths:
        agents_dir = agents_root

    monkeypatch.setattr("deerflow.config.paths.get_paths", lambda: _FakePaths())

    assert curator._collect_referenced_skill_names() == {"alpha", "beta", "gamma"}