"""Tests for the ``POST /api/skills/reconcile-custom`` adoption flow.

Covers the scenario where a user drops a skill directory onto disk by hand
(bypassing the upload API) and then calls the reconcile route so it shows up
in the UI.
"""

from __future__ import annotations

import errno
import json
import os
from pathlib import Path
from types import SimpleNamespace

import pytest
from fastapi import HTTPException

from app.gateway.routers import skills as skills_router
from deerflow.community.aio_sandbox import aio_sandbox_provider as aio_provider_module
from deerflow.community.aio_sandbox.remote_backend import RemoteSandboxBackend
from deerflow.config import paths as paths_module
from deerflow.config.extensions_config import get_extensions_config, reset_extensions_config
from deerflow.persistence.skills import MemorySkillStore
from deerflow.sandbox import tools as sandbox_tools
from deerflow.sandbox.local.local_sandbox import LocalSandbox, PathMapping
from deerflow.sandbox.skill_view import (
    SKILL_SUPPORT_MANIFEST,
    SKILL_VIEW_MARKER,
    refresh_skill_sandbox_view,
)
from deerflow.skills.storage.local_skill_storage import LocalSkillStorage
from deerflow.tools.builtins import skill_tools


def _write_skill_on_disk(root: Path, name: str, description: str = "from disk") -> None:
    """Drop a valid SKILL.md into ``<root>/custom/<name>/`` like a manual unzip would."""
    skill_dir = root / "custom" / name
    skill_dir.mkdir(parents=True, exist_ok=True)
    (skill_dir / "SKILL.md").write_text(
        f"---\nname: {name}\ndescription: {description}\n---\n# {name}\n",
        encoding="utf-8",
    )


def _fake_request(user_id: str | None = "user-a") -> SimpleNamespace:
    user = SimpleNamespace(id=user_id, system_role="user") if user_id else None
    return SimpleNamespace(state=SimpleNamespace(user=user))


@pytest.fixture
def skills_root(tmp_path: Path) -> Path:
    root = tmp_path / "skills"
    (root / "public").mkdir(parents=True)
    (root / "custom").mkdir(parents=True)
    return root


@pytest.fixture
def patched_storage(
    skills_root: Path, monkeypatch: pytest.MonkeyPatch
) -> LocalSkillStorage:
    """Force the router to use a LocalSkillStorage rooted at ``skills_root``."""
    storage = LocalSkillStorage(host_path=str(skills_root))
    monkeypatch.setattr(
        skills_router, "get_or_new_skill_storage", lambda **_: storage
    )

    async def _noop_refresh() -> None:
        return None

    monkeypatch.setattr(
        skills_router, "refresh_skills_system_prompt_cache_async", _noop_refresh
    )
    return storage


@pytest.mark.asyncio
async def test_reconcile_adopts_orphan_custom_skill(
    skills_root: Path, patched_storage: LocalSkillStorage
):
    _write_skill_on_disk(skills_root, "manually-dropped")
    store = MemorySkillStore()

    result = await skills_router.reconcile_custom_skills(
        request=_fake_request("user-a"), config=None, skill_store=store
    )

    assert result.adopted == ["manually-dropped"]
    assert result.already_registered == []

    record = await store.get_any("manually-dropped")
    assert record is not None
    assert record["owner_user_id"] == "user-a"
    assert record["published"] is False


@pytest.mark.asyncio
async def test_reconcile_skips_already_registered(
    skills_root: Path, patched_storage: LocalSkillStorage
):
    _write_skill_on_disk(skills_root, "already-known")
    store = MemorySkillStore()
    # Pre-existing ownership row — should be left alone.
    await store.create(
        {"name": "already-known", "owner_user_id": "user-b", "published": True}
    )

    result = await skills_router.reconcile_custom_skills(
        request=_fake_request("user-a"), config=None, skill_store=store
    )

    assert result.adopted == []
    assert result.already_registered == ["already-known"]

    # Ownership must not be overwritten.
    record = await store.get_any("already-known")
    assert record is not None
    assert record["owner_user_id"] == "user-b"
    assert record["published"] is True


@pytest.mark.asyncio
async def test_reconcile_mixed_adopts_only_orphans(
    skills_root: Path, patched_storage: LocalSkillStorage
):
    _write_skill_on_disk(skills_root, "orphan-one")
    _write_skill_on_disk(skills_root, "orphan-two")
    _write_skill_on_disk(skills_root, "registered")
    store = MemorySkillStore()
    await store.create(
        {"name": "registered", "owner_user_id": "user-a", "published": False}
    )

    result = await skills_router.reconcile_custom_skills(
        request=_fake_request("user-a"), config=None, skill_store=store
    )

    assert sorted(result.adopted) == ["orphan-one", "orphan-two"]
    assert result.already_registered == ["registered"]


@pytest.mark.asyncio
async def test_reconcile_with_empty_custom_dir_is_noop(
    skills_root: Path, patched_storage: LocalSkillStorage
):
    store = MemorySkillStore()

    result = await skills_router.reconcile_custom_skills(
        request=_fake_request("user-a"), config=None, skill_store=store
    )

    assert result.adopted == []
    assert result.already_registered == []


def _write_availability_skill(skills_root: Path) -> Path:
    skill_file = skills_root / "public" / "non-canonical-directory" / "SKILL.md"
    skill_file.parent.mkdir(parents=True, exist_ok=True)
    skill_file.write_text(
        "---\nname: demo-skill\ndescription: Availability test skill\n---\n\n# Distinct skill body\n",
        encoding="utf-8",
    )
    return skill_file


def _availability_request(*, role: str, user_id: str = "test-user") -> SimpleNamespace:
    return SimpleNamespace(
        state=SimpleNamespace(user=SimpleNamespace(id=user_id, system_role=role)),
    )


@pytest.fixture
def isolated_skill_environment(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
    config_path = tmp_path / "extensions_config.json"
    availability_skills_root = tmp_path / "availability-skills"
    skill_file = _write_availability_skill(availability_skills_root)
    monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_path))
    monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path / "home"))
    monkeypatch.setattr(paths_module, "_paths", None)
    reset_extensions_config()
    try:
        yield config_path, LocalSkillStorage(host_path=str(availability_skills_root)), skill_file
    finally:
        reset_extensions_config()
        monkeypatch.setattr(paths_module, "_paths", None)


@pytest.mark.asyncio
async def test_update_public_skill_is_admin_only_and_preserves_raw_config(
    isolated_skill_environment,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config_path, storage, _skill_file = isolated_skill_environment
    original = {
        "mcpServers": {
            "preserved": {
                "type": "stdio",
                "command": "$MCP_COMMAND",
                "args": ["--flag"],
                "futureMcpField": {"keep": True},
            },
        },
        "skills": {
            "demo-skill": {
                "enabled": True,
                "display": {
                    "enabled": True,
                    "mode": "list",
                    "resultPath": "hits",
                    "futureDisplayField": "keep",
                },
                "futureSkillField": ["keep"],
            },
            "other-skill": {"enabled": False, "unknown": 7},
        },
        "futureTopLevel": {"placeholder": "$DO_NOT_RESOLVE", "keep": True},
    }
    config_path.write_text(json.dumps(original, ensure_ascii=False), encoding="utf-8")
    monkeypatch.setenv("MCP_COMMAND", "resolved-secret-command")
    monkeypatch.setattr(skills_router, "get_or_new_skill_storage", lambda **_kwargs: storage)

    refresh_calls: list[bool] = []
    catalog_refresh_calls: list[object] = []
    enabled_row_updates: list[tuple[str, bool]] = []

    async def _refresh() -> None:
        refresh_calls.append(True)

    async def _refresh_catalog(config) -> None:
        catalog_refresh_calls.append(config)

    async def _set_enabled(name: str, enabled: bool) -> bool:
        enabled_row_updates.append((name, enabled))
        return True

    monkeypatch.setattr(skills_router, "refresh_skills_system_prompt_cache_async", _refresh)
    monkeypatch.setattr(skills_router, "refresh_skill_catalog_projection", _refresh_catalog)
    monkeypatch.setattr(skills_router, "set_skill_catalog_enabled", _set_enabled)
    replace_calls: list[tuple[object, object]] = []
    real_replace = os.replace

    def _replace(source, destination) -> None:
        replace_calls.append((source, destination))
        real_replace(source, destination)

    monkeypatch.setattr(skills_router.os, "replace", _replace)

    before = config_path.read_bytes()
    with pytest.raises(HTTPException) as forbidden:
        await skills_router.update_skill(
            "demo-skill",
            skills_router.SkillUpdateRequest(enabled=False),
            _availability_request(role="user"),
            config=None,
        )
    assert forbidden.value.status_code == 403
    assert config_path.read_bytes() == before
    assert refresh_calls == []
    assert enabled_row_updates == []

    response = await skills_router.update_skill(
        "demo-skill",
        skills_router.SkillUpdateRequest(enabled=False),
        _availability_request(role="admin"),
        config=None,
    )

    persisted = json.loads(config_path.read_text(encoding="utf-8"))
    assert response.enabled is False
    assert persisted["skills"]["demo-skill"]["enabled"] is False
    assert persisted["skills"]["demo-skill"]["display"] == original["skills"]["demo-skill"]["display"]
    assert persisted["skills"]["demo-skill"]["futureSkillField"] == ["keep"]
    assert persisted["mcpServers"] == original["mcpServers"]
    assert persisted["futureTopLevel"] == original["futureTopLevel"]
    assert persisted["mcpServers"]["preserved"]["command"] == "$MCP_COMMAND"
    assert get_extensions_config().skills["demo-skill"].enabled is False
    assert refresh_calls == [True]
    assert enabled_row_updates == [("demo-skill", False)]
    assert catalog_refresh_calls == []
    config_replaces = [item for item in replace_calls if os.fspath(item[1]) == os.fspath(config_path)]
    assert len(config_replaces) == 1

    # A second request must re-read the raw document rather than writing a
    # stale singleton snapshot.
    persisted["concurrentUnknownField"] = {"survives": True}
    config_path.write_text(json.dumps(persisted, ensure_ascii=False), encoding="utf-8")
    response = await skills_router.update_skill(
        "demo-skill",
        skills_router.SkillUpdateRequest(enabled=True),
        _availability_request(role="admin"),
        config=None,
    )
    persisted = json.loads(config_path.read_text(encoding="utf-8"))
    assert response.enabled is True
    assert persisted["skills"]["demo-skill"]["enabled"] is True
    assert persisted["concurrentUnknownField"] == {"survives": True}
    assert refresh_calls == [True, True]
    assert enabled_row_updates == [("demo-skill", False), ("demo-skill", True)]
    assert catalog_refresh_calls == []
    config_replaces = [item for item in replace_calls if os.fspath(item[1]) == os.fspath(config_path)]
    assert len(config_replaces) == 2


@pytest.mark.asyncio
async def test_update_custom_skill_allows_owner_but_rejects_other_users(
    isolated_skill_environment,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config_path, storage, _skill_file = isolated_skill_environment
    skills_root = storage.get_skills_root_path()
    _write_skill_on_disk(skills_root, "owned-skill")
    _write_skill_on_disk(skills_root, "other-skill")
    _write_skill_on_disk(skills_root, "orphan-skill")
    config_path.write_text(
        json.dumps(
            {
                "skills": {
                    "owned-skill": {"enabled": True},
                    "other-skill": {"enabled": True},
                    "orphan-skill": {"enabled": True},
                }
            }
        ),
        encoding="utf-8",
    )
    monkeypatch.setattr(skills_router, "get_or_new_skill_storage", lambda **_kwargs: storage)

    refresh_calls: list[bool] = []

    async def _refresh() -> None:
        refresh_calls.append(True)

    monkeypatch.setattr(skills_router, "refresh_skills_system_prompt_cache_async", _refresh)

    skill_store = MemorySkillStore()
    await skill_store.create(
        {"name": "owned-skill", "owner_user_id": "user-a", "published": False}
    )
    await skill_store.create(
        {"name": "other-skill", "owner_user_id": "user-b", "published": True}
    )

    response = await skills_router.update_skill(
        "owned-skill",
        skills_router.SkillUpdateRequest(enabled=False),
        _availability_request(role="user", user_id="user-a"),
        config=None,
        skill_store=skill_store,
    )
    assert response.enabled is False
    persisted = json.loads(config_path.read_text(encoding="utf-8"))
    assert persisted["skills"]["owned-skill"]["enabled"] is False
    assert refresh_calls == [True]

    response = await skills_router.update_skill(
        "owned-skill",
        skills_router.SkillUpdateRequest(enabled=True),
        _availability_request(role="user", user_id="user-a"),
        config=None,
        skill_store=skill_store,
    )
    assert response.enabled is True
    persisted = json.loads(config_path.read_text(encoding="utf-8"))
    assert persisted["skills"]["owned-skill"]["enabled"] is True
    assert refresh_calls == [True, True]

    with pytest.raises(HTTPException) as unauthenticated:
        await skills_router.update_skill(
            "owned-skill",
            skills_router.SkillUpdateRequest(enabled=False),
            SimpleNamespace(state=SimpleNamespace(user=None)),
            config=None,
            skill_store=skill_store,
        )
    assert unauthenticated.value.status_code == 403

    for forbidden_name in ("other-skill", "orphan-skill"):
        with pytest.raises(HTTPException) as forbidden:
            await skills_router.update_skill(
                forbidden_name,
                skills_router.SkillUpdateRequest(enabled=False),
                _availability_request(role="user", user_id="user-a"),
                config=None,
                skill_store=skill_store,
            )
        assert forbidden.value.status_code == 403

    persisted = json.loads(config_path.read_text(encoding="utf-8"))
    assert persisted["skills"]["other-skill"]["enabled"] is True
    assert persisted["skills"]["orphan-skill"]["enabled"] is True
    assert refresh_calls == [True, True]

    admin_response = await skills_router.update_skill(
        "orphan-skill",
        skills_router.SkillUpdateRequest(enabled=False),
        _availability_request(role="admin", user_id="admin-user"),
        config=None,
        skill_store=skill_store,
    )
    assert admin_response.enabled is False
    persisted = json.loads(config_path.read_text(encoding="utf-8"))
    assert persisted["skills"]["orphan-skill"]["enabled"] is False
    assert refresh_calls == [True, True, True]


@pytest.mark.asyncio
async def test_update_skill_falls_back_for_bind_mount_ebusy(
    isolated_skill_environment,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config_path, storage, _skill_file = isolated_skill_environment
    original = {
        "mcpServers": {"server": {"command": "$SECRET", "future": "keep"}},
        "skills": {"demo-skill": {"enabled": True, "future": {"keep": True}}},
        "futureTopLevel": ["keep"],
    }
    config_path.write_text(json.dumps(original, ensure_ascii=False), encoding="utf-8")
    monkeypatch.setattr(skills_router, "get_or_new_skill_storage", lambda **_kwargs: storage)

    refresh_calls: list[bool] = []

    async def _refresh() -> None:
        refresh_calls.append(True)

    monkeypatch.setattr(skills_router, "refresh_skills_system_prompt_cache_async", _refresh)

    def _busy_replace(_source, _destination) -> None:
        raise OSError(errno.EBUSY, "bind mount")

    monkeypatch.setattr(skills_router.os, "replace", _busy_replace)

    response = await skills_router.update_skill(
        "demo-skill",
        skills_router.SkillUpdateRequest(enabled=False),
        _availability_request(role="admin"),
        config=None,
    )

    persisted = json.loads(config_path.read_text(encoding="utf-8"))
    assert response.enabled is False
    assert persisted["skills"]["demo-skill"]["enabled"] is False
    assert persisted["skills"]["demo-skill"]["future"] == {"keep": True}
    assert persisted["mcpServers"] == original["mcpServers"]
    assert persisted["futureTopLevel"] == ["keep"]
    assert refresh_calls == [True]
    assert list(config_path.parent.glob("tmp*")) == []


@pytest.mark.asyncio
async def test_display_and_delete_state_writes_preserve_other_raw_config(
    isolated_skill_environment,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config_path, storage, _skill_file = isolated_skill_environment
    original = {
        "mcpServers": {"server": {"command": "$SECRET", "future": "keep"}},
        "skills": {
            "demo-skill": {
                "enabled": False,
                "display": {"enabled": False, "mode": "none"},
                "futureSkillField": "keep",
            },
            "other-skill": {"enabled": True, "unknown": 7},
        },
        "futureTopLevel": {"keep": True},
    }
    config_path.write_text(json.dumps(original, ensure_ascii=False), encoding="utf-8")
    monkeypatch.setattr(skills_router, "get_or_new_skill_storage", lambda **_kwargs: storage)

    async def _noop_upsert(*_args, **_kwargs) -> None:
        return None

    async def _no_adapter(_skill_name: str):
        return None

    async def _noop_refresh() -> None:
        return None

    monkeypatch.setattr(skills_router, "_upsert_skill_display_adapter", _noop_upsert)
    monkeypatch.setattr(skills_router, "_get_skill_display_adapter", _no_adapter)
    monkeypatch.setattr(skills_router, "refresh_skills_system_prompt_cache_async", _noop_refresh)

    await skills_router.update_skill_display(
        "demo-skill",
        skills_router.SkillDisplayUpdateRequest(
            display=skills_router.SkillDisplayConfig(enabled=True, mode="list", resultPath="hits"),
        ),
        config=None,
    )

    persisted = json.loads(config_path.read_text(encoding="utf-8"))
    assert persisted["skills"]["demo-skill"]["enabled"] is False
    assert persisted["skills"]["demo-skill"]["futureSkillField"] == "keep"
    assert persisted["skills"]["demo-skill"]["display"]["resultPath"] == "hits"
    assert persisted["mcpServers"] == original["mcpServers"]
    assert persisted["futureTopLevel"] == {"keep": True}

    skills_router._remove_skill_state_config("demo-skill")
    persisted = json.loads(config_path.read_text(encoding="utf-8"))
    assert "demo-skill" not in persisted["skills"]
    assert persisted["skills"]["other-skill"] == original["skills"]["other-skill"]
    assert persisted["mcpServers"] == original["mcpServers"]
    assert persisted["futureTopLevel"] == {"keep": True}


def test_skill_list_and_view_hide_offline_skill_and_restore_it(
    isolated_skill_environment,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    config_path, storage, skill_file = isolated_skill_environment
    config_path.write_text(
        json.dumps({"mcpServers": {}, "skills": {"demo-skill": {"enabled": False}}}),
        encoding="utf-8",
    )

    import deerflow.skills.storage as storage_module

    monkeypatch.setattr(storage_module, "get_or_new_skill_storage", lambda **_kwargs: storage)
    monkeypatch.setattr(skill_tools, "_load_owned_skill_names", lambda _user_id: None)
    monkeypatch.setattr(skill_tools, "_resolve_effective_user_id", lambda: "test-user")

    offline_list = json.loads(skill_tools.skill_list_impl(agent_id=None))
    assert all(item["name"] != "demo-skill" for item in offline_list)
    assert skill_tools.skill_view_impl("demo-skill", agent_id=None) == "Skill 'demo-skill' not found."

    config_path.write_text(
        json.dumps({"mcpServers": {}, "skills": {"demo-skill": {"enabled": True}}}),
        encoding="utf-8",
    )
    online_list = json.loads(skill_tools.skill_list_impl(agent_id=None))
    assert any(item["name"] == "demo-skill" for item in online_list)
    assert skill_tools.skill_view_impl("demo-skill", agent_id=None) == skill_file.read_text(encoding="utf-8")


def test_skill_path_access_gate_uses_discovered_directory_and_fails_closed(
    isolated_skill_environment,
) -> None:
    config_path, storage, _skill_file = isolated_skill_environment
    skill_path = "/mnt/skills/public/non-canonical-directory/bin/run.py"
    config_path.write_text(
        json.dumps({"mcpServers": {}, "skills": {"demo-skill": {"enabled": False}}}),
        encoding="utf-8",
    )

    with pytest.raises(skill_tools.SkillPathUnavailableError):
        skill_tools.require_enabled_skill_path(skill_path, storage=storage)
    with pytest.raises(skill_tools.SkillPathUnavailableError):
        skill_tools.require_enabled_skill_path("/mnt/skills", storage=storage)
    with pytest.raises(skill_tools.SkillPathUnavailableError):
        skill_tools.require_enabled_skill_path("/mnt/skills/public", storage=storage)

    config_path.write_text(
        json.dumps({"mcpServers": {}, "skills": {"demo-skill": {"enabled": True}}}),
        encoding="utf-8",
    )
    skill = skill_tools.require_enabled_skill_path(skill_path, storage=storage)
    assert skill.name == "demo-skill"


@pytest.mark.parametrize(
    ("tool", "args"),
    [
        (sandbox_tools.bash_tool, (None, "run skill", "python /mnt/skills/public/demo-skill/run.py")),
        (sandbox_tools.ls_tool, (None, "list skill", "/mnt/skills/public/demo-skill")),
        (sandbox_tools.glob_tool, (None, "find skill files", "**/*.py", "/mnt/skills/public/demo-skill")),
        (sandbox_tools.grep_tool, (None, "search skill", "needle", "/mnt/skills/public/demo-skill")),
        (sandbox_tools.read_file_tool, (None, "read skill", "/mnt/skills/public/demo-skill/SKILL.md")),
    ],
)
def test_skill_path_gate_runs_before_local_or_remote_sandbox_dispatch(
    tool,
    args,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    sandbox_calls: list[bool] = []

    def _deny(_path: str) -> None:
        raise skill_tools.SkillPathUnavailableError("offline")

    def _unexpected_sandbox(_runtime):
        sandbox_calls.append(True)
        raise AssertionError("sandbox dispatch must not occur")

    monkeypatch.setattr(sandbox_tools, "_require_enabled_skill_path", _deny)
    monkeypatch.setattr(sandbox_tools, "ensure_sandbox_initialized", _unexpected_sandbox)

    result = tool.func(*args)

    assert result.startswith("Error:")
    assert sandbox_calls == []


def _write_named_skill(skills_root: Path, directory: str, name: str, body: str) -> Path:
    skill_dir = skills_root / "public" / directory
    skill_dir.mkdir(parents=True, exist_ok=True)
    (skill_dir / "SKILL.md").write_text(
        f"---\nname: {name}\ndescription: Capability test\n---\n\n{body}\n",
        encoding="utf-8",
    )
    return skill_dir


def test_online_skill_view_removes_offline_files_and_keeps_declared_internal_dependency(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    runtime_root = tmp_path / "runtime"
    monkeypatch.setenv("DEER_FLOW_HOME", str(runtime_root))
    monkeypatch.setattr(paths_module, "_paths", None)
    skills_root = tmp_path / "skills"
    online_dir = _write_named_skill(skills_root, "online-directory", "online-skill", "online")
    offline_dir = _write_named_skill(skills_root, "offline-directory", "offline-skill", "offline")
    (online_dir / "online-secret.txt").write_text("ONLINE-CONTENT", encoding="utf-8")
    (offline_dir / "offline-secret.txt").write_text("OFFLINE-CONTENT", encoding="utf-8")

    support_dir = skills_root / "public" / "task-report-lib"
    support_dir.mkdir(parents=True)
    (support_dir / "convert_lib.py").write_text("VALUE = 'dependency-present'\n", encoding="utf-8")
    (support_dir / SKILL_SUPPORT_MANIFEST).write_text(
        json.dumps({"required_by": ["online-skill"]}),
        encoding="utf-8",
    )

    config_path = tmp_path / "extensions_config.json"
    config_path.write_text(
        json.dumps(
            {
                "skills": {
                    "online-skill": {"enabled": True},
                    "offline-skill": {"enabled": False},
                },
            }
        ),
        encoding="utf-8",
    )
    monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_path))
    storage = LocalSkillStorage(host_path=str(skills_root))

    view = refresh_skill_sandbox_view(storage=storage)

    assert view.root.parent == runtime_root.resolve()
    assert skills_root.resolve() not in view.root.parents
    assert (view.root / "public" / "online-directory" / "online-secret.txt").read_text(encoding="utf-8") == "ONLINE-CONTENT"
    assert not (view.root / "public" / "offline-directory").exists()
    assert (view.root / "public" / "task-report-lib" / "convert_lib.py").is_file()
    assert (view.root / SKILL_VIEW_MARKER).read_text(encoding="utf-8") == view.marker_content
    assert (offline_dir / "offline-secret.txt").read_text(encoding="utf-8") == "OFFLINE-CONTENT"

    config_path.write_text(
        json.dumps(
            {
                "skills": {
                    "online-skill": {"enabled": False},
                    "offline-skill": {"enabled": False},
                },
            }
        ),
        encoding="utf-8",
    )
    refreshed = refresh_skill_sandbox_view(storage=storage)
    assert not (refreshed.root / "public" / "online-directory").exists()
    assert not (refreshed.root / "public" / "task-report-lib").exists()


def test_online_skill_view_fails_closed_when_extensions_config_is_invalid(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    runtime_root = tmp_path / "runtime"
    monkeypatch.setenv("DEER_FLOW_HOME", str(runtime_root))
    monkeypatch.setattr(paths_module, "_paths", None)
    skills_root = tmp_path / "skills"
    _write_named_skill(skills_root, "online", "online-skill", "online")
    config_path = tmp_path / "extensions_config.json"
    config_path.write_text(
        json.dumps({"skills": {"online-skill": {"enabled": True}}}),
        encoding="utf-8",
    )
    monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_path))
    storage = LocalSkillStorage(host_path=str(skills_root))

    view = refresh_skill_sandbox_view(storage=storage)
    assert (view.root / "public" / "online" / "SKILL.md").is_file()

    config_path.write_text("{", encoding="utf-8")
    with pytest.raises(ValueError, match="not valid JSON"):
        refresh_skill_sandbox_view(storage=storage)

    assert not (view.root / SKILL_VIEW_MARKER).exists()
    assert not any(view.root.iterdir())


def test_local_sandbox_view_blocks_env_and_dynamic_python_bypasses(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path / "runtime"))
    monkeypatch.setattr(paths_module, "_paths", None)
    skills_root = tmp_path / "skills"
    online_dir = _write_named_skill(skills_root, "online", "online-skill", "online")
    offline_dir = _write_named_skill(skills_root, "offline", "offline-skill", "offline")
    (online_dir / "value.txt").write_text("ONLINE-VALUE", encoding="utf-8")
    (offline_dir / "value.txt").write_text("OFFLINE-SECRET", encoding="utf-8")

    config_path = tmp_path / "extensions_config.json"
    config_path.write_text(
        json.dumps(
            {
                "skills": {
                    "online-skill": {"enabled": True},
                    "offline-skill": {"enabled": False},
                },
            }
        ),
        encoding="utf-8",
    )
    monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_path))
    view = refresh_skill_sandbox_view(storage=LocalSkillStorage(host_path=str(skills_root)))
    sandbox = LocalSandbox(
        "capability-test",
        path_mappings=[PathMapping("/mnt/skills", str(view.root), read_only=True)],
    )

    monkeypatch.setattr(sandbox_tools, "_get_skills_host_path", lambda: str(view.root))
    enabled_command = sandbox_tools.replace_virtual_paths_in_command(
        'python -c "from pathlib import Path; print(Path(\'/mnt/skills/public/online/value.txt\').read_text())"',
        None,
    )
    enabled_result = sandbox.execute_command(enabled_command)
    assert "ONLINE-VALUE" in enabled_result

    env_bypass = sandbox.execute_command(
        'python -c "import os; from pathlib import Path; '
        "os.environ['SKILL_ROOT']='/'+'mnt'+'/'+'skills'; "
        "print(Path(os.environ['SKILL_ROOT'], 'public', 'offline', 'value.txt').read_text())\""
    )
    dynamic_bypass = sandbox.execute_command(
        'python -c "from pathlib import Path; '
        "path=Path('/'+'mnt')/'skills'/'public'/'offline'/'value.txt'; "
        'print(path.read_text())"'
    )

    assert "OFFLINE-SECRET" not in env_bypass
    assert "OFFLINE-SECRET" not in dynamic_bypass
    assert "Exit Code:" in env_bypass
    assert "Exit Code:" in dynamic_bypass


def test_local_sandbox_command_environment_hides_backend_skill_locations(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv("DEER_FLOW_SKILLS_PATH", "/canonical/skills")
    monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", "/canonical/extensions.json")
    monkeypatch.setenv("DEER_FLOW_GATEWAY_PORT", "8001")
    monkeypatch.setenv("PYTHONPATH", "/backend/checkout")
    monkeypatch.setenv("SAFE_SKILL_API_TOKEN", "kept")

    environment = LocalSandbox._model_command_environment()

    assert "DEER_FLOW_SKILLS_PATH" not in environment
    assert "DEER_FLOW_EXTENSIONS_CONFIG_PATH" not in environment
    assert "PYTHONPATH" not in environment
    assert environment["DEER_FLOW_GATEWAY_PORT"] == "8001"
    assert environment["SAFE_SKILL_API_TOKEN"] == "kept"


def test_aio_sandbox_mounts_and_attests_only_the_online_view(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    view_root = tmp_path / "runtime" / ".sandbox-online-skills"
    view_root.mkdir(parents=True)
    view = SimpleNamespace(root=view_root, marker_content='{"token":"expected"}')
    config = SimpleNamespace(skills=SimpleNamespace(container_path="/mnt/skills"))
    monkeypatch.setattr(aio_provider_module, "refresh_skill_sandbox_view", lambda: view)
    monkeypatch.setattr(aio_provider_module, "get_app_config", lambda: config)
    monkeypatch.setenv("DEER_FLOW_HOST_BASE_DIR", "/host/runtime")

    mount = aio_provider_module.AioSandboxProvider._get_skills_mount()
    assert mount is not None
    assert mount[0].replace("\\", "/").endswith("/host/runtime/.sandbox-online-skills")
    assert mount[1:] == ("/mnt/skills", True)

    good_sandbox = SimpleNamespace(read_file=lambda path: view.marker_content)
    aio_provider_module.AioSandboxProvider._assert_online_skill_view(good_sandbox)
    bad_sandbox = SimpleNamespace(read_file=lambda path: "Error: missing")
    with pytest.raises(RuntimeError, match="online-only skills view"):
        aio_provider_module.AioSandboxProvider._assert_online_skill_view(bad_sandbox)


def test_remote_provisioner_receives_security_sensitive_extra_mounts(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    captured: dict[str, object] = {}

    class _Response:
        def raise_for_status(self) -> None:
            return None

        def json(self) -> dict[str, str]:
            return {"sandbox_url": "http://sandbox:8080"}

    def _post(url: str, *, json: dict, timeout: int):
        captured.update({"url": url, "json": json, "timeout": timeout})
        return _Response()

    monkeypatch.setattr("deerflow.community.aio_sandbox.remote_backend.requests.post", _post)
    backend = RemoteSandboxBackend("http://provisioner:8002")
    backend.create(
        "thread-1",
        "sandbox-1",
        extra_mounts=[("/host/runtime/.sandbox-online-skills", "/mnt/skills", True)],
    )

    assert captured["json"]["extra_mounts"] == [
        {
            "host_path": "/host/runtime/.sandbox-online-skills",
            "container_path": "/mnt/skills",
            "read_only": True,
        }
    ]