from io import StringIO
from unittest.mock import patch

import pytest
from rich.console import Console

from cli import ChatConsole
from hermes_cli.skills_hub import do_check, do_install, do_list, do_update, handle_skills_slash


class _DummyLockFile:
    def __init__(self, installed):
        self._installed = installed

    def list_installed(self):
        return self._installed


@pytest.fixture()
def hub_env(monkeypatch, tmp_path):
    """Set up isolated hub directory paths and return (monkeypatch, tmp_path)."""
    import tools.skills_hub as hub

    hub_dir = tmp_path / "skills" / ".hub"
    monkeypatch.setattr(hub, "SKILLS_DIR", tmp_path / "skills")
    monkeypatch.setattr(hub, "HUB_DIR", hub_dir)
    monkeypatch.setattr(hub, "LOCK_FILE", hub_dir / "lock.json")
    monkeypatch.setattr(hub, "QUARANTINE_DIR", hub_dir / "quarantine")
    monkeypatch.setattr(hub, "AUDIT_LOG", hub_dir / "audit.log")
    monkeypatch.setattr(hub, "TAPS_FILE", hub_dir / "taps.json")
    monkeypatch.setattr(hub, "INDEX_CACHE_DIR", hub_dir / "index-cache")

    return hub_dir


# ---------------------------------------------------------------------------
# Fixtures for common skill setups
# ---------------------------------------------------------------------------

_HUB_ENTRY = {"name": "hub-skill", "source": "github", "trust_level": "community"}

_ALL_THREE_SKILLS = [
    {"name": "hub-skill", "category": "x", "description": "hub"},
    {"name": "builtin-skill", "category": "x", "description": "builtin"},
    {"name": "local-skill", "category": "x", "description": "local"},
]

_BUILTIN_MANIFEST = {"builtin-skill": "abc123"}


@pytest.fixture()
def three_source_env(monkeypatch, hub_env):
    """Populate hub/builtin/local skills for source-classification tests."""
    import tools.skills_hub as hub
    from tools import skills_sync
    from tools import skills_tool

    monkeypatch.setattr(hub, "HubLockFile", lambda: _DummyLockFile([_HUB_ENTRY]))
    monkeypatch.setattr(skills_tool, "find_all_skills", lambda **_kwargs: list(_ALL_THREE_SKILLS))
    monkeypatch.setattr(skills_sync, "_read_manifest", lambda: dict(_BUILTIN_MANIFEST))

    return hub_env


def _capture(source_filter: str = "all") -> str:
    """Run do_list into a string buffer and return the output."""
    sink = StringIO()
    console = Console(file=sink, force_terminal=False, color_system=None)
    do_list(source_filter=source_filter, console=console)
    return sink.getvalue()


def _capture_check(monkeypatch, results, name=None) -> str:
    import tools.skills_hub_install as hub_install

    sink = StringIO()
    console = Console(file=sink, force_terminal=False, color_system=None)
    monkeypatch.setattr(hub_install, "check_for_skill_updates", lambda **_kwargs: results)
    do_check(name=name, console=console)
    return sink.getvalue()


def _capture_update(monkeypatch, results) -> tuple[str, list[tuple[str, str, bool]]]:
    import tools.skills_hub as hub
    import tools.skills_hub_install as hub_install
    import hermes_cli.skills_hub as cli_hub

    sink = StringIO()
    console = Console(file=sink, force_terminal=False, color_system=None)
    installs = []

    monkeypatch.setattr(hub_install, "check_for_skill_updates", lambda **_kwargs: results)
    monkeypatch.setattr(hub, "HubLockFile", lambda: type("L", (), {
        "get_installed": lambda self, name: {"install_path": "category/" + name}
    })())
    monkeypatch.setattr(cli_hub, "do_install", lambda identifier, category="", force=False, console=None, source_id=None: installs.append((identifier, category, force)))

    do_update(console=console)
    return sink.getvalue(), installs


# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------




def test_do_list_platform_env_is_ignored(three_source_env, monkeypatch):
    """`hermes skills list` reads the active profile's config via
    HERMES_HOME (swapped by -p), so it must NOT pass a platform arg to
    ``get_disabled_skill_names`` — otherwise per-platform overrides
    would silently leak in from HERMES_PLATFORM env."""
    from agent import skill_utils

    seen = {}

    def _fake(platform=None):
        seen["platform"] = platform
        return set()

    monkeypatch.setattr(skill_utils, "get_disabled_skill_names", _fake)
    _capture()

    assert seen["platform"] is None


# ---------------------------------------------------------------------------
# Cross-registry hijack regression tests
#
# An update must never change a skill's source registry. Skill names are not
# namespaced across registries, so an unconstrained name resolve can install a
# different author's same-named skill over the user's files.
# ---------------------------------------------------------------------------




def test_check_for_skill_updates_does_not_fall_back_across_registries(tmp_path, monkeypatch):
    """An entry whose source has no adapter reports `unavailable`.

    Previously `candidate_sources ... or sources` fell back to every source, so
    a same-named skill in another registry could satisfy the fetch and be
    reported as this entry's update -- the step that preceded the overwrite.
    The foreign source here returns a *valid* bundle with a different hash, so
    the old code reports `update_available` (sourced from the wrong registry)
    while the fixed code reports `unavailable`.
    """
    from tools.skills_hub_install import check_for_skill_updates

    class _ForeignBundle:
        name = "reddit"
        files = {"SKILL.md": "# a different author's reddit skill"}
        source = "skills.sh"
        identifier = "skills-sh/someone-else/reddit"
        trust_level = "community"
        metadata: dict = {}

    class _ForeignSource:
        """skills-sh adapter; must NOT be consulted for a clawhub-locked entry."""

        def source_id(self):
            return "skills-sh"

        def fetch(self, identifier):
            return _ForeignBundle()

        def inspect(self, identifier):
            return _ForeignBundle()

    from tools import skills_hub as hub
    monkeypatch.setattr(hub, "SKILLS_DIR", tmp_path)
    (tmp_path / "reddit").mkdir()
    lock = _DummyLockFile([
        {"name": "reddit", "identifier": "reddit", "source": "clawhub",
         "install_path": "reddit", "content_hash": "hash-of-the-clawhub-copy"},
    ])

    results = check_for_skill_updates(
        lock=lock,  # type: ignore[arg-type]  # duck-typed double, matches _DummyLockFile usage above
        sources=[_ForeignSource()],  # type: ignore[list-item]
    )

    assert len(results) == 1
    assert results[0]["source"] == "clawhub", "provenance must be preserved"
    assert results[0]["status"] == "unavailable", (
        "a clawhub-locked skill must not be matched against a skills-sh bundle; "
        "reporting update_available here is the cross-registry hijack"
    )
    assert "bundle" not in results[0], "must not carry a foreign registry's bundle"


def test_resolve_does_not_pair_catalog_meta_with_foreign_same_name_bundle():
    """Inspect metadata from one registry must not ship with another registry's files.

    skills.sh can inspect ``owner/repo/skills/skillopt`` while ClawHub used to
    fetch by last path segment ``skillopt`` and return a different author's
    SKILL.md. The header then showed the requested identifier and the preview
    showed the wrong skill.
    """
    from hermes_cli.skills_hub import _resolve_source_meta_and_bundle
    from tools.skills_hub_models import SkillBundle, SkillMeta

    class CatalogSource:
        def inspect(self, identifier):
            return SkillMeta(
                name="skillopt",
                description="kanban-based pipelines",
                source="skills.sh",
                identifier="skills-sh/latipun7/agent-skill-collections/skills/skillopt",
                trust_level="community",
            )

        def fetch(self, identifier):
            return None

    class ForeignSlugSource:
        def inspect(self, identifier):
            return SkillMeta(
                name="skillopt",
                description="Train, evaluate, and improve Agent skill files",
                source="clawhub",
                identifier="skillopt",
                trust_level="community",
            )

        def fetch(self, identifier):
            return SkillBundle(
                name="skillopt",
                files={"SKILL.md": "# Train, evaluate, and improve Agent skill files\n"},
                source="clawhub",
                identifier="skillopt",
                trust_level="community",
            )

    meta, bundle, matched = _resolve_source_meta_and_bundle(
        "latipun7/agent-skill-collections/skills/skillopt",
        [CatalogSource(), ForeignSlugSource()],
    )

    assert bundle is not None
    assert bundle.source == "clawhub"
    assert meta is not None
    assert meta.source == "clawhub"
    assert meta.identifier == "skillopt"
    assert "kanban-based" not in (meta.description or "")
    assert matched is not None
    assert matched.__class__ is ForeignSlugSource


def test_resolve_keeps_catalog_meta_when_later_sources_do_not_fetch():
    from hermes_cli.skills_hub import _resolve_source_meta_and_bundle
    from tools.skills_hub_models import SkillMeta

    class CatalogSource:
        def inspect(self, identifier):
            return SkillMeta(
                name="skillopt",
                description="kanban-based pipelines",
                source="skills.sh",
                identifier="skills-sh/latipun7/agent-skill-collections/skills/skillopt",
                trust_level="community",
            )

        def fetch(self, identifier):
            return None

    class QuietSource:
        def inspect(self, identifier):
            return None

        def fetch(self, identifier):
            return None

    meta, bundle, matched = _resolve_source_meta_and_bundle(
        "latipun7/agent-skill-collections/skills/skillopt",
        [CatalogSource(), QuietSource()],
    )

    assert bundle is None
    assert meta is not None
    assert meta.source == "skills.sh"
    assert meta.identifier.endswith("latipun7/agent-skill-collections/skills/skillopt")
    assert matched is not None
    assert matched.__class__ is CatalogSource


def test_inspect_reuses_one_ssrf_safe_client_for_metadata_and_bundle(monkeypatch, tmp_path):
    """A preview's (and an install's) sequential resolver calls must share one guarded connection pool."""
    import hermes_cli.skills_hub as cli_hub
    import tools.skills_hub as hub
    import tools.skills_hub_search as search
    import tools.skills_hub_clawhub as clawhub
    from tools.skills_hub_models import SkillBundle, SkillMeta

    clients = []
    client_kwargs = []

    class Response:
        status_code = 200
        text = "ok"
        content = b"ok"
        headers = {}

        def json(self):
            return {"skills": []}

    class Client:
        def __enter__(self):
            return self

        def __exit__(self, *_args):
            return False

        def get(self, *_args, **_kwargs):
            return Response()

    class Source:
        def source_id(self):
            return "github"

        def inspect(self, _identifier):
            hub._guarded_http_get("https://example.com/metadata")
            # The Hermes-index fetch must ride the same pool (no cache → real GET).
            assert search._load_hermes_index() == {"skills": []}
            return SkillMeta("example", "metadata", "test", "example/id", "community")

        def fetch(self, _identifier):
            hub._guarded_http_get("https://example.com/SKILL.md")
            return SkillBundle("example", {"SKILL.md": "# Example"}, "test", "example/id", "community")

    def create_client(**kwargs):
        client_kwargs.append(kwargs)
        client = Client()
        clients.append(client)
        return client

    monkeypatch.setattr(cli_hub, "_sources", lambda: [Source()])
    monkeypatch.setattr(hub, "create_ssrf_safe_client", create_client, raising=False)
    monkeypatch.setattr(hub, "is_safe_url", lambda _url: True)
    monkeypatch.setattr(hub, "check_website_access", lambda _url: None)
    monkeypatch.setattr(search, "_hermes_index_cache_file", lambda: tmp_path / "hermes-index.json")
    monkeypatch.setattr(search.httpx, "get", lambda *_a, **_k: pytest.fail("index fetch bypassed the pool"))

    result = cli_hub.inspect_skill("example/id")

    assert result is not None
    assert len(clients) == 1
    # The pooled client keeps the one-shot default timeout and never auto-follows redirects.
    assert client_kwargs[0]["timeout"] == hub._DEFAULT_HTTP_TIMEOUT
    assert client_kwargs[0]["follow_redirects"] is False

    # Install runs the same resolve + fetch sequence (plus the per-file fan-out) and must pool too.
    clients.clear()
    monkeypatch.setattr(cli_hub, "_resolve_url_bundle_name", lambda *_a, **_k: False)
    cli_hub.do_install("example/id", console=Console(file=StringIO()))
    assert len(clients) == 1




# ---------------------------------------------------------------------------
# UrlSource-specific install paths: --name override, interactive prompts,
# non-interactive error, existing-category scan.
# ---------------------------------------------------------------------------


def _make_url_bundle_fetcher(name="", awaiting_name=True, url="https://example.com/SKILL.md"):
    """Return a fake source that simulates ``UrlSource.fetch`` for a
    URL-sourced skill whose name hasn't been auto-resolved."""

    class _UrlSource:
        def inspect(self, identifier):
            return type("Meta", (), {
                "extra": {"url": url, "awaiting_name": awaiting_name},
                "identifier": url,
                "name": name,
                "path": name,
            })()

        def fetch(self, identifier):
            return type("Bundle", (), {
                "name": name,
                "files": {"SKILL.md": "---\ndescription: ok\n---\n# body\n"},
                "source": "url",
                "identifier": url,
                "trust_level": "community",
                "metadata": {"url": url, "awaiting_name": awaiting_name},
            })()

    return _UrlSource


def _install_mocks(monkeypatch, tmp_path, source_factory, category_hint=""):
    """Wire the minimum set of monkeypatches for a do_install dry run."""
    import tools.skills_hub as hub
    import tools.skills_hub_install as hub_install
    import tools.skills_hub_search as hub_search
    import tools.skills_guard as guard

    q_path = tmp_path / "skills" / ".hub" / "quarantine" / "pending"
    q_path.mkdir(parents=True)

    install_calls: list = []

    def _install_from_quarantine(q, name, category, bundle, result):
        install_calls.append({"name": name, "category": category})
        install_dir = tmp_path / "skills" / (f"{category}/" if category else "") / name
        install_dir.mkdir(parents=True, exist_ok=True)
        return install_dir

    monkeypatch.setattr(hub, "ensure_hub_dirs", lambda: None)
    monkeypatch.setattr(hub_search, "create_source_router", lambda auth: [source_factory()])
    monkeypatch.setattr(hub_install, "quarantine_bundle", lambda bundle: q_path)
    monkeypatch.setattr(hub_install, "install_from_quarantine", _install_from_quarantine)
    monkeypatch.setattr(
        hub, "HubLockFile",
        lambda: type("Lock", (), {"get_installed": lambda self, n: None})(),
    )
    monkeypatch.setattr(
        guard, "scan_skill",
        lambda skill_path, source="community": guard.ScanResult(
            skill_name="pending", source=source, trust_level="community", verdict="safe",
        ),
    )
    monkeypatch.setattr(guard, "format_scan_report", lambda result: "scan ok")
    monkeypatch.setattr(guard, "should_allow_install", lambda result, force=False: (True, "ok"))
    return install_calls






# ── _existing_categories ────────────────────────────────────────────────────






# ---------------------------------------------------------------------------
# browse_skills — dedup by identifier, not name
# ---------------------------------------------------------------------------




# ---------------------------------------------------------------------------
# Regression: full identifier must be recoverable from `hermes skills search`
# even when the slug is too long to fit the terminal width (issue #33674).
# ---------------------------------------------------------------------------

# A real browse-sh-style slug whose trailing -XXXXXX hash matters for install
_LONG_SLUG = "browse-sh/weather.gov/get-forecast-1uezib"

_LONG_RESULT = type("R", (), {
    "name": "get-forecast",
    "description": "Fetch the forecast",
    "source": "browse-sh",
    "trust_level": "community",
    "identifier": _LONG_SLUG,
})()


def test_do_search_json_flag_emits_full_identifiers(capsys):
    """`--json` must print a parseable array with full identifiers and skip the table."""
    from hermes_cli.skills_hub import do_search

    sink = StringIO()
    console = Console(file=sink, force_terminal=False, color_system=None, width=40)

    with patch("tools.skills_hub_search.unified_search", return_value=[_LONG_RESULT]), \
         patch("tools.skills_hub_search.create_source_router", return_value={}), \
         patch("tools.skills_hub_github.GitHubAuth"):
        do_search("weather", console=console, as_json=True)

    # JSON goes to stdout via print(), not the Rich console sink.
    captured = capsys.readouterr().out
    import json as _json
    payload = _json.loads(captured)
    assert isinstance(payload, list) and len(payload) == 1
    assert payload[0]["identifier"] == _LONG_SLUG
    assert payload[0]["name"] == "get-forecast"
    assert payload[0]["source"] == "browse-sh"
    # Table render must be suppressed — sink should be empty (no "Searching for:" header).
    assert "Searching for:" not in sink.getvalue()



# ---------------------------------------------------------------------------
# Local-edit protection in do_update (ported from paperclipai/paperclip#10978)
# ---------------------------------------------------------------------------


def _update_env(monkeypatch, tmp_path, *, edit_after_install: bool):
    """Install a fake hub skill on disk, optionally edit it, and wire mocks.

    Returns (console_sink, installs_list).
    """
    import hermes_cli.skills_hub as cli_hub
    import tools.skills_hub as hub
    import tools.skills_hub_install as hub_install
    from tools.skills_guard import content_hash

    skills_dir = tmp_path / "skills"
    skill_dir = skills_dir / "category" / "hub-skill"
    skill_dir.mkdir(parents=True)
    (skill_dir / "SKILL.md").write_text("# hub-skill\noriginal\n")

    recorded = content_hash(skill_dir)
    if edit_after_install:
        (skill_dir / "SKILL.md").write_text("# hub-skill\nuser edited\n")

    monkeypatch.setattr(hub, "SKILLS_DIR", skills_dir)
    monkeypatch.setattr(hub_install, "check_for_skill_updates", lambda **_kwargs: [{
        "name": "hub-skill",
        "identifier": "someone/hub-skill",
        "source": "github",
        "status": "update_available",
    }])
    monkeypatch.setattr(hub, "HubLockFile", lambda: type("L", (), {
        "get_installed": lambda self, name: {
            "install_path": "category/hub-skill",
            "content_hash": recorded,
        }
    })())

    installs = []
    monkeypatch.setattr(
        cli_hub, "do_install",
        lambda identifier, category="", force=False, console=None, source_id=None:
            installs.append(identifier),
    )

    sink = StringIO()
    console = Console(file=sink, force_terminal=False, color_system=None)
    return console, sink, installs


def test_do_update_skips_locally_edited_skill(monkeypatch, tmp_path):
    """A hub skill whose on-disk hash drifted from the lockfile is skipped."""
    console, sink, installs = _update_env(monkeypatch, tmp_path, edit_after_install=True)

    do_update(console=console)

    assert installs == []
    out = sink.getvalue()
    assert "local edits" in out
    assert "--force" in out


def test_do_update_force_overwrites_local_edits(monkeypatch, tmp_path):
    """--force restores the destructive replace for edited skills."""
    console, sink, installs = _update_env(monkeypatch, tmp_path, edit_after_install=True)

    do_update(console=console, force=True)

    assert installs == ["someone/hub-skill"]
    assert "local edits" not in sink.getvalue()


def test_do_update_unmodified_skill_updates_normally(monkeypatch, tmp_path):
    """No local drift -> the update proceeds without --force."""
    console, sink, installs = _update_env(monkeypatch, tmp_path, edit_after_install=False)

    do_update(console=console)

    assert installs == ["someone/hub-skill"]
    assert "Updated 1 skill(s)" in sink.getvalue()


# ---------------------------------------------------------------------------
# Stale index entry messages (#3259)
# ---------------------------------------------------------------------------


def _stale_env(monkeypatch):
    """do_install where the index has metadata but the files are gone (404)."""
    import hermes_cli.skills_hub as cli_hub
    import tools.skills_hub as hub

    class StaleSource:
        def source_id(self):
            return "skills-sh"

    meta = type("Meta", (), {"identifier": "skills-sh/org/gone-skill"})()
    monkeypatch.setattr(hub, "ensure_hub_dirs", lambda: None)
    monkeypatch.setattr(cli_hub, "_sources", lambda: [StaleSource()])
    monkeypatch.setattr(
        cli_hub, "_resolve_source_meta_and_bundle",
        lambda identifier, sources: (meta, None, sources[0]))
    sink = StringIO()
    console = Console(file=sink, force_terminal=False, color_system=None)
    return console, sink


def test_do_install_stale_index_names_the_problem(monkeypatch):
    """Index hit + missing files reads as a stale entry, not a typo (#3259)."""
    from hermes_cli.skills_hub import do_install

    console, sink = _stale_env(monkeypatch)
    do_install("skills-sh/org/gone-skill", console=console, skip_confirm=True)

    out = sink.getvalue()
    assert "Stale index entry" in out
    assert "skills-sh" in out
    assert "Could not fetch" not in out


def test_fetch_failure_names_rejected_github_credential():
    """A credential GitHub refused is named, never reported as a stale index entry (#98725)."""
    from types import SimpleNamespace
    from hermes_cli.skills_hub import _print_fetch_failure

    src = SimpleNamespace(is_rate_limited=False, auth=SimpleNamespace(rejected=["GITHUB_TOKEN/GH_TOKEN"]),
                          source_id=lambda: "hermes-index")
    sink = StringIO()
    console = Console(file=sink, force_terminal=False, color_system=None, width=200)
    _print_fetch_failure(console, [src], "o/r/s", meta=object(), source=src)

    out = sink.getvalue()
    assert "rejected GITHUB_TOKEN/GH_TOKEN" in out
    assert "Stale index entry" not in out


@pytest.mark.parametrize("meta_hit", [False, True])
def test_do_install_generic_when_no_index_hit_or_rate_limited(monkeypatch, meta_hit):
    """No index hit — or a throttled fetch that only *looks* like a stale entry — keeps the
    generic message (plus the rate-limit hint), never the stale-entry verdict."""
    import hermes_cli.skills_hub as cli_hub
    import tools.skills_hub as hub
    from hermes_cli.skills_hub import do_install

    class ThrottledSource:
        is_rate_limited = meta_hit

        def source_id(self):
            return "skills-sh"

    meta = type("Meta", (), {"identifier": "skills-sh/org/gone-skill"})() if meta_hit else None
    src = ThrottledSource()
    monkeypatch.setattr(hub, "ensure_hub_dirs", lambda: None)
    monkeypatch.setattr(cli_hub, "_sources", lambda: [src])
    monkeypatch.setattr(
        cli_hub, "_resolve_source_meta_and_bundle",
        lambda identifier, sources: (meta, None, src if meta_hit else None))
    sink = StringIO()
    console = Console(file=sink, force_terminal=False, color_system=None)
    do_install("skills-sh/org/gone-skill", console=console, skip_confirm=True)

    out = sink.getvalue()
    assert "Could not download" in out
    assert "Stale index entry" not in out
    assert ("rate limit" in out) is meta_hit


# ---------------------------------------------------------------------------
# Short-name resolution by identifier + failure exit code
# ---------------------------------------------------------------------------


def _meta(name, identifier, source="clawhub"):
    from tools.skills_hub_models import SkillMeta

    return SkillMeta(name=name, description="", source=source, identifier=identifier,
                     trust_level="community")


def _sink_console():
    sink = StringIO()
    return Console(file=sink, force_terminal=False, color_system=None), sink


def _stub_search(monkeypatch, results):
    """unified_search is late-bound inside _resolve_short_name, so patch its module."""
    import tools.skills_hub_search as search

    monkeypatch.setattr(search, "unified_search", lambda *args, **kwargs: list(results))


@pytest.mark.parametrize("identifier", [
    "computer-use", "NousResearch/hermes-agent/skills/productivity/pdf",
    "skills-sh/nousresearch/hermes-agent/google-workspace", "blender-bpy-enhanced"])
def test_install_by_name_resolves_to_the_skill_it_names(monkeypatch, tmp_path, identifier):
    """`hermes skills install <bundled skill>` failed every time: the name resolved to a stranger's
    same-named hub skill or an ambiguity table, and this repo's own copy was rescanned as community
    content and refused. It now makes the shipped skill active and fetches nothing. A hub slug behind
    a prettified catalog title ("Blender Bpy Enhanced") resolves too, where it answered "No exact match"
    while printing that very slug as the suggestion."""
    import hermes_cli.skills_hub as cli_hub

    if identifier == "blender-bpy-enhanced":
        _stub_search(monkeypatch, [_meta("Blender Bpy Enhanced", "@emergencescience/blender-bpy-enhanced")])
        console, sink = _sink_console()
        assert cli_hub._resolve_short_name(identifier, [], console) == "@emergencescience/blender-bpy-enhanced"
        assert "No exact match" not in sink.getvalue()
        return
    from tools import skills_sync

    import tools.skills_hub as hub
    from tools import skill_usage
    from tools.skills_hub import HubLockFile

    # hub_env's undo leaves its resolved paths as real globals; drop them so the hub follows the profile.
    for stale in [n for n in vars(hub) if n.isupper() and n.endswith(("_DIR", "_FILE", "_LOG"))]:
        monkeypatch.delitem(vars(hub), stale)
    monkeypatch.setattr(cli_hub, "_sources", lambda: pytest.fail("a bundled skill must not hit the hub"))
    name = identifier.rsplit("/", 1)[-1]
    # The curator pruned it (prune_builtins on): suppressed from re-seeding and recorded archived.
    monkeypatch.setattr(skill_usage, "_prune_builtins_enabled", lambda: True)
    skill_usage._toggle_suppressed_name(name, add=True)
    skill_usage.save_usage({name: {"state": skill_usage.STATE_ARCHIVED}})
    # A symlinked category must not carry the restore out of the skills tree (the hub's guard).
    category = skill_usage._skills_dir() / dict(skills_sync._discover_bundled_skills(
        skills_sync._get_bundled_dir()))[name].parent.name
    (outside := tmp_path / "outside").mkdir()
    category.parent.mkdir(parents=True, exist_ok=True)
    category.symlink_to(outside, target_is_directory=True)
    console, sink = _sink_console()
    assert cli_hub.do_install(identifier, console=console, skip_confirm=True) is False
    assert not any(outside.iterdir())
    category.unlink()

    assert cli_hub.do_install(identifier, console=console, skip_confirm=True) is True
    assert "Restored built-in skill" in sink.getvalue()
    assert name in skills_sync._read_manifest()
    assert any(skills_sync.read_skill_name(md, "") == name for md in skills_sync._iter_active_skill_mds())
    assert name not in skill_usage.read_suppressed_names()
    assert skill_usage.load_usage()[name]["state"] == skill_usage.STATE_ACTIVE
    # Already active now: a no-op the user owns, not a failure (exit 0, no reinstall).
    assert cli_hub.do_install(identifier, console=console, skip_confirm=True) is None
    # A hub skill that took the name is reported as such (with the way back), never as the built-in.
    stranger = skill_usage._skills_dir() / "hub-installs" / name
    stranger.mkdir(parents=True)
    (stranger / "SKILL.md").write_text(f"---\nname: {name}\n---\nstranger\n")
    HubLockFile().record_install(name=name, source="skills-sh", identifier=f"skills-sh/x/y/{name}",
                                 trust_level="community", scan_verdict="safe", skill_hash="h",
                                 install_path=f"hub-installs/{name}", files=["SKILL.md"])
    console, sink = _sink_console()
    assert cli_hub.do_install(identifier, console=console, skip_confirm=True) is False
    # One line, subject and remedy together: the Desktop toast shows only a failed action's last 3 lines.
    error = next(line for line in sink.getvalue().splitlines() if line.startswith("Error:"))
    assert f"'{name}' is a built-in skill" in error and f"`hermes skills uninstall {name}`" in error
    assert "already available" not in sink.getvalue()
    assert (stranger / "SKILL.md").read_text().endswith("stranger\n")


def test_built_in_restore_survives_leftovers_and_records_every_outcome_as_bundled(monkeypatch, tmp_path):
    """Post-fix telemetry still showed `llm-wiki` failing as source=hub, filesystem_error: the
    restore copied with copytree onto a dest dir left behind without its SKILL.md (EEXIST on every
    retry), refused a skill already present behind a symlinked category the loader follows, and a
    failed restore was labelled a hub install."""
    import hermes_cli.skills_hub as cli_hub
    import tools.skills_hub as hub
    from tools import skill_usage, skills_sync

    for stale in [n for n in vars(hub) if n.isupper() and n.endswith(("_DIR", "_FILE", "_LOG"))]:
        monkeypatch.delitem(vars(hub), stale)
    monkeypatch.setattr(cli_hub, "_sources", lambda: pytest.fail("a bundled skill must not hit the hub"))
    rows = []
    monkeypatch.setattr("hermes_cli.observability.shared_metrics_events.record_extension_install",
                        lambda **row: rows.append((row["source"], row["outcome"])))
    src = dict(skills_sync._discover_bundled_skills(skills_sync._get_bundled_dir()))["llm-wiki"]
    dest = skill_usage._skills_dir() / src.parent.name / "llm-wiki"
    console, _sink = _sink_console()

    (dest / "wiki").mkdir(parents=True)  # stranded: user data kept, SKILL.md gone
    (dest / "wiki" / "index.md").write_text("mine\n")
    assert cli_hub.do_install("llm-wiki", console=console, skip_confirm=True) is True
    assert (dest / "SKILL.md").is_file() and (dest / "wiki" / "index.md").read_text() == "mine\n"

    real = tmp_path / "real-category"
    dest.parent.rename(real)
    dest.parent.symlink_to(real, target_is_directory=True)
    assert cli_hub.do_install("llm-wiki", console=console, skip_confirm=True) is None  # present, no-op

    (real / "llm-wiki" / "SKILL.md").unlink()  # missing behind the symlink: refused, and still bundled
    assert cli_hub.do_install("llm-wiki", console=console, skip_confirm=True) is False
    assert rows == [("bundled", "success"), ("bundled", "failed")]


def test_built_in_restore_into_a_leftover_dir_never_writes_through_links_or_over_user_files(tmp_path):
    """The leftover-tolerant restore copied with dirs_exist_ok: it followed a symlinked file out of
    skills/ (clobbering the target), overwrote a user-edited file, and called a dir holding another
    skill's SKILL.md the built-in "already available"."""
    from tools import skill_usage, skills_sync
    from tools.skills_sync_bundled_ops import ensure_bundled_skill

    src = dict(skills_sync._discover_bundled_skills(skills_sync._get_bundled_dir()))["arxiv"]
    dest = skill_usage._skills_dir() / src.parent.name / "arxiv"
    victim = tmp_path / "victim.txt"
    victim.write_text("USER SECRET\n")
    (dest / "scripts").mkdir(parents=True)
    (dest / "scripts" / "search_arxiv.py").symlink_to(victim)
    assert ensure_bundled_skill("arxiv")["action"] == "not_restored"
    assert victim.read_text() == "USER SECRET\n" and not (dest / "SKILL.md").exists()

    (dest / "scripts" / "search_arxiv.py").unlink()
    (dest / "scripts" / "search_arxiv.py").write_text("# my local edits\n")
    assert ensure_bundled_skill("arxiv")["action"] == "restored"
    assert (dest / "scripts" / "search_arxiv.py").read_text() == "# my local edits\n"
    assert (dest / "SKILL.md").read_bytes() == (src / "SKILL.md").read_bytes()

    (dest / "SKILL.md").write_text("---\nname: my-own-thing\ndescription: x\n---\nmine\n")
    assert ensure_bundled_skill("arxiv")["action"] == "not_restored"
    assert "my-own-thing" in (dest / "SKILL.md").read_text()