from __future__ import annotations

import json
import shutil
import tomllib
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from threading import Event

import pytest

from opensquilla.engine.context import load_context_files
from opensquilla.migration.openclaw import (
    MigrationOptions,
    OpenClawMigrator,
    _ApplyRollback,
)
from opensquilla.onboarding.config_store import load_config, persist_config
from opensquilla.provider.selector import build_provider
from opensquilla.skills.loader import SkillLoader
from opensquilla.skills.types import SkillLayer


def _make_source(root: Path) -> Path:
    source = root / ".openclaw"
    workspace = source / "workspace"
    (workspace / "memory").mkdir(parents=True)
    (workspace / "skills" / "demo").mkdir(parents=True)
    (workspace / "tts").mkdir(parents=True)
    (source / "credentials").mkdir(parents=True)

    (workspace / "SOUL.md").write_text("openclaw soul\n", encoding="utf-8")
    (workspace / "AGENTS.md").write_text("agent rules\n", encoding="utf-8")
    (workspace / "USER.md").write_text("user profile\n", encoding="utf-8")
    (workspace / "MEMORY.md").write_text("core memory\n", encoding="utf-8")
    (workspace / "memory" / "2026-05-10.md").write_text("daily note\n", encoding="utf-8")
    (workspace / "skills" / "demo" / "SKILL.md").write_text(
        "---\nname: demo\ndescription: Demo skill\n---\nDemo body\n",
        encoding="utf-8",
    )
    (workspace / "tts" / "voice.txt").write_text("voice asset\n", encoding="utf-8")
    (source / "exec-approvals.json").write_text(
        json.dumps({"allow": ["^git status$", "^pytest "]}, ensure_ascii=False),
        encoding="utf-8",
    )
    (source / "openclaw.json").write_text(
        json.dumps(
            {
                "agents": {
                    "defaults": {
                        "model": "deepseek-chat",
                        "timeoutSeconds": 500,
                    }
                },
                "mcp": {
                    "servers": {
                        "demo-mcp": {
                            "command": "node",
                            "args": ["server.js"],
                            "env": {"DEMO": "1"},
                        }
                    }
                },
                "tools": {"exec": {"timeoutSec": 45}},
                "messages": {
                    "telegram": {"defaultChatId": "12345"},
                    "discord": {"defaultChannelId": "C999"},
                    "tts": {"provider": "elevenlabs", "elevenlabs": {"voice": "voice-1"}},
                },
            },
            ensure_ascii=False,
        ),
        encoding="utf-8",
    )
    return source


def test_failed_rollback_retains_before_images_and_reports_recovery_path(
    tmp_path: Path,
    monkeypatch,
) -> None:
    target = tmp_path / "target.txt"
    target.write_text("before\n", encoding="utf-8")
    rollback = _ApplyRollback()
    backup_root = rollback._backup_root

    def fail_restore(cls, image) -> None:
        del cls, image
        raise OSError("forced restore failure")

    monkeypatch.setattr(_ApplyRollback, "_restore", classmethod(fail_restore))

    try:
        with pytest.raises(OSError, match="before-image backup retained") as caught:
            with rollback:
                rollback.capture(target)
                target.write_text("after\n", encoding="utf-8")
                raise RuntimeError("forced apply failure")

        assert str(backup_root) in str(caught.value)
        assert backup_root.is_dir()
        assert (backup_root / "0").read_text(encoding="utf-8") == "before\n"
    finally:
        shutil.rmtree(backup_root, ignore_errors=True)


def test_late_failure_preserves_concurrent_settings_save(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    home = tmp_path / "opensquilla-home"
    config_path = tmp_path / "config.toml"
    config_path.write_text("debug = false\n", encoding="utf-8")
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    report_started = Event()
    settings_saved = Event()

    def fail_report_write(self) -> None:
        del self
        report_started.set()
        if not settings_saved.wait(timeout=5):
            raise AssertionError("settings save did not complete")
        raise RuntimeError("forced late failure")

    monkeypatch.setattr(OpenClawMigrator, "_write_report_files", fail_report_write)
    migrator = OpenClawMigrator(
        MigrationOptions(source=source, config_path=config_path, apply=True)
    )

    with ThreadPoolExecutor(max_workers=1) as pool:
        migration = pool.submit(migrator.migrate)
        assert report_started.wait(timeout=5)
        try:
            settings = load_config(config_path, persist_migrations=False)
            settings.debug = True
            persist_config(settings, path=config_path, backup=False)
        finally:
            settings_saved.set()

        with pytest.raises(RuntimeError, match="forced late failure"):
            migration.result(timeout=5)

    persisted = tomllib.loads(config_path.read_text(encoding="utf-8"))
    assert persisted["debug"] is True
    assert "llm" not in persisted


def test_dry_run_does_not_write_targets(tmp_path: Path, monkeypatch) -> None:
    source = _make_source(tmp_path)
    home = tmp_path / "opensquilla-home"
    config_path = tmp_path / "config.toml"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    report = OpenClawMigrator(
        MigrationOptions(source=source, config_path=config_path, apply=False)
    ).migrate()

    assert report["apply"] is False
    assert any(item["status"] == "planned" for item in report["items"])
    assert not config_path.exists()
    assert not (home / "workspace" / "SOUL.md").exists()
    assert not (home / "skills" / "openclaw-imports").exists()


def test_dry_run_does_not_upgrade_legacy_target_config(tmp_path: Path, monkeypatch) -> None:
    source = _make_source(tmp_path)
    home = tmp_path / "opensquilla-home"
    config_path = tmp_path / "config.toml"
    fixture = (
        Path(__file__).parent
        / "fixtures"
        / "homes"
        / "cli-0.1"
        / "config.toml"
    )
    shutil.copy2(fixture, config_path)
    original = config_path.read_bytes()
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    OpenClawMigrator(
        MigrationOptions(source=source, config_path=config_path, apply=False)
    ).migrate()

    assert config_path.read_bytes() == original
    assert list(tmp_path.glob("config.toml.backup.*")) == []


def test_apply_defers_legacy_target_config_upgrade_to_final_persist(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    home = tmp_path / "opensquilla-home"
    config_path = tmp_path / "config.toml"
    original = (
        f"workspace_dir = {json.dumps(str(home / 'workspace'))}\n"
        f"state_dir = {json.dumps(str(home / 'state'))}\n"
        "\n[llm_ensemble]\n"
        "proposer_timeout_seconds = 120.0\n"
        "aggregator_timeout_seconds = 120.0\n"
    )
    config_path.write_text(original, encoding="utf-8")
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    OpenClawMigrator(
        MigrationOptions(
            source=source,
            config_path=config_path,
            apply=True,
            preset="user-data",
            # Skill import is the only selected user-data operation that
            # intentionally changes config; exclude it so this test isolates
            # the deferred target-schema persist.
            exclude=("skills", "shared-skills"),
        )
    ).migrate()

    persisted = tomllib.loads(config_path.read_text(encoding="utf-8"))
    assert persisted["config_version"] == 1
    assert persisted["llm_ensemble"]["proposer_timeout_seconds"] == 3600.0
    assert persisted["llm_ensemble"]["aggregator_timeout_seconds"] == 3600.0
    backups = list(tmp_path.glob("config.toml.backup.*"))
    assert len(backups) == 1
    assert backups[0].read_text(encoding="utf-8") == original


def test_apply_migrates_workspace_skills_config_and_allowlist(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    home = tmp_path / "opensquilla-home"
    config_path = tmp_path / "config.toml"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    report = OpenClawMigrator(
        MigrationOptions(source=source, config_path=config_path, apply=True)
    ).migrate()

    workspace = home / "workspace"
    assert (workspace / "SOUL.md").read_text(encoding="utf-8") == "opensquilla soul\n"
    assert (workspace / "AGENTS.md").read_text(encoding="utf-8") == "agent rules\n"
    memory = (workspace / "MEMORY.md").read_text(encoding="utf-8")
    assert "core memory" in memory
    assert "daily note" in memory

    skill_file = home / "skills" / "openclaw-imports" / "demo" / "SKILL.md"
    assert skill_file.read_text(encoding="utf-8").startswith("---\nname: demo")

    config = tomllib.loads(config_path.read_text(encoding="utf-8"))
    assert config["llm"]["model"] == "deepseek-chat"
    assert config["mcp"]["enabled"] is True
    assert config["mcp"]["servers"][0]["name"] == "demo-mcp"
    assert str(home / "skills" / "openclaw-imports") in config["skills"]["extra_dirs"]
    assert config["agent_runtime_timeout_seconds"] == 500

    env_text = (home / ".env").read_text(encoding="utf-8")
    assert "OPENSQUILLA_SAFE_BIN_ALLOW=^git status$,^pytest " in env_text
    assert (Path(report["output_dir"]) / "report.json").is_file()
    assert (Path(report["output_dir"]) / "summary.md").is_file()


def test_explicit_state_dir_normalizes_default_workspace_paths(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    home = tmp_path / "opensquilla-home"
    config_path = tmp_path / "config.toml"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))
    config_path.write_text(
        'workspace_dir = "'
        + str(Path.home() / ".opensquilla" / "workspace").replace("\\", "\\\\")
        + '"\nstate_dir = "'
        + str(Path.home() / ".opensquilla" / "state").replace("\\", "\\\\")
        + '"\n',
        encoding="utf-8",
    )

    report = OpenClawMigrator(
        MigrationOptions(source=source, config_path=config_path, apply=True)
    ).migrate()

    workspace = home / "workspace"
    assert (workspace / "SOUL.md").read_text(encoding="utf-8") == "opensquilla soul\n"
    assert not any("target exists" in item["reason"] for item in report["items"])
    config = tomllib.loads(config_path.read_text(encoding="utf-8"))
    assert config["workspace_dir"] == str(home / "workspace")
    assert config["state_dir"] == str(home / "state")


def test_secrets_are_opt_in_and_reports_are_redacted(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    (source / ".env").write_text("DEEPSEEK_API_KEY=sk-secret-value\n", encoding="utf-8")
    home = tmp_path / "opensquilla-home"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    no_secret_report = OpenClawMigrator(
        MigrationOptions(source=source, config_path=tmp_path / "no-secret.toml", apply=True)
    ).migrate()
    assert "sk-secret-value" not in json.dumps(no_secret_report)
    assert "DEEPSEEK_API_KEY=sk-secret-value" not in (home / ".env").read_text(
        encoding="utf-8"
    )

    secret_report = OpenClawMigrator(
        MigrationOptions(
            source=source,
            config_path=tmp_path / "with-secret.toml",
            apply=True,
            migrate_secrets=True,
        )
    ).migrate()

    assert "DEEPSEEK_API_KEY=sk-secret-value" in (home / ".env").read_text(encoding="utf-8")
    encoded = json.dumps(secret_report)
    assert "sk-secret-value" not in encoded
    assert "[redacted]" in encoded


def test_migrate_secrets_can_create_supported_channels(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    (source / ".env").write_text(
        "TELEGRAM_BOT_TOKEN=tg-secret\nDISCORD_BOT_TOKEN=discord-secret\n",
        encoding="utf-8",
    )
    home = tmp_path / "opensquilla-home"
    config_path = tmp_path / "config.toml"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    report = OpenClawMigrator(
        MigrationOptions(
            source=source,
            config_path=config_path,
            apply=True,
            migrate_secrets=True,
        )
    ).migrate()

    config = tomllib.loads(config_path.read_text(encoding="utf-8"))
    channels = {entry["type"]: entry for entry in config["channels"]["channels"]}
    assert channels["telegram"]["token"] == "tg-secret"
    assert channels["telegram"]["default_chat_id"] == "12345"
    assert channels["discord"]["token"] == "discord-secret"
    assert channels["discord"]["default_channel_id"] == "C999"
    assert "tg-secret" not in json.dumps(report)
    assert "discord-secret" not in json.dumps(report)


def test_tts_assets_are_copied_and_tts_config_is_archived(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    home = tmp_path / "opensquilla-home"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    report = OpenClawMigrator(
        MigrationOptions(source=source, config_path=tmp_path / "config.toml", apply=True)
    ).migrate()

    assert (home / "tts" / "voice.txt").read_text(encoding="utf-8") == "voice asset\n"
    archive = Path(report["output_dir"]) / "archive" / "tts-config.json"
    assert archive.is_file()
    assert any(
        item["kind"] == "tts-config" and item["status"] == "archived"
        for item in report["items"]
    )


def test_unmapped_openclaw_config_is_archived(tmp_path: Path, monkeypatch) -> None:
    source = _make_source(tmp_path)
    config = json.loads((source / "openclaw.json").read_text(encoding="utf-8"))
    config["plugins"] = {"enabled": True, "entries": [{"id": "custom"}]}
    (source / "openclaw.json").write_text(json.dumps(config), encoding="utf-8")
    home = tmp_path / "opensquilla-home"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    report = OpenClawMigrator(
        MigrationOptions(source=source, config_path=tmp_path / "config.toml", apply=True)
    ).migrate()

    archive = Path(report["output_dir"]) / "archive" / "plugins-config.json"
    assert archive.is_file()
    assert any(
        item["kind"] == "plugins-config" and item["status"] == "archived"
        for item in report["items"]
    )


def test_archive_redacts_secret_values_by_key_name(tmp_path: Path, monkeypatch) -> None:
    source = _make_source(tmp_path)
    config = json.loads((source / "openclaw.json").read_text(encoding="utf-8"))
    config["plugins"] = {
        "entries": [
            {
                "id": "custom",
                "config": {
                    "apiKey": "opaque-value-123",
                    "nested": {"webhookToken": "opaque-value-456"},
                },
            }
        ]
    }
    (source / "openclaw.json").write_text(json.dumps(config), encoding="utf-8")
    home = tmp_path / "opensquilla-home"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    report = OpenClawMigrator(
        MigrationOptions(source=source, config_path=tmp_path / "config.toml", apply=True)
    ).migrate()

    archive_text = (
        Path(report["output_dir"]) / "archive" / "plugins-config.json"
    ).read_text(encoding="utf-8")
    assert "opaque-value-123" not in archive_text
    assert "opaque-value-456" not in archive_text
    assert archive_text.count("[redacted]") == 2


def test_user_data_preset_skips_runtime_config_and_archives(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    config = json.loads((source / "openclaw.json").read_text(encoding="utf-8"))
    config["plugins"] = {"entries": [{"id": "custom"}]}
    (source / "openclaw.json").write_text(json.dumps(config), encoding="utf-8")
    home = tmp_path / "opensquilla-home"
    config_path = tmp_path / "config.toml"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    report = OpenClawMigrator(
        MigrationOptions(
            source=source,
            config_path=config_path,
            apply=True,
            preset="user-data",
        )
    ).migrate()

    assert (home / "workspace" / "SOUL.md").is_file()
    assert (home / "skills" / "openclaw-imports" / "demo" / "SKILL.md").is_file()
    persisted = tomllib.loads(config_path.read_text(encoding="utf-8"))
    assert str(home / "skills" / "openclaw-imports") in persisted["skills"]["extra_dirs"]
    # The user-data preset skips model config entirely; with sparse
    # persistence the llm section is not written at all.
    assert persisted.get("llm", {}).get("model") != "deepseek-chat"
    assert not (Path(report["output_dir"]) / "archive" / "plugins-config.json").exists()
    assert not any(item["kind"] == "model-config" for item in report["items"])


def test_granular_archive_option_archives_only_selected_config(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    config = json.loads((source / "openclaw.json").read_text(encoding="utf-8"))
    config["plugins"] = {"entries": [{"id": "custom"}]}
    config["cron"] = {"jobs": [{"id": "nightly"}]}
    (source / "openclaw.json").write_text(json.dumps(config), encoding="utf-8")
    home = tmp_path / "opensquilla-home"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    report = OpenClawMigrator(
        MigrationOptions(
            source=source,
            config_path=tmp_path / "config.toml",
            apply=True,
            preset="user-data",
            include=("plugins-config",),
        )
    ).migrate()

    archive_dir = Path(report["output_dir"]) / "archive"
    assert (archive_dir / "plugins-config.json").is_file()
    assert not (archive_dir / "cron-config.json").exists()
    assert any(
        item["kind"] == "plugins-config" and item["status"] == "archived"
        for item in report["items"]
    )
    assert not any(item["kind"] == "cron-jobs" for item in report["items"])


def test_imported_skill_without_frontmatter_is_reported_not_loadable(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    workspace = source / "workspace"
    bad_skill = workspace / "skills" / "no-frontmatter"
    bad_skill.mkdir(parents=True)
    (bad_skill / "SKILL.md").write_text("Missing frontmatter body\n", encoding="utf-8")
    home = tmp_path / "opensquilla-home"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    report = OpenClawMigrator(
        MigrationOptions(
            source=source,
            config_path=tmp_path / "config.toml",
            apply=True,
            preset="user-data",
        )
    ).migrate()

    item = next(
        item
        for item in report["items"]
        if item["source"] and item["source"].endswith("no-frontmatter")
    )
    assert item["status"] == "migrated"
    assert item["details"]["opensquilla_loadable"] is False
    assert "missing YAML frontmatter" in item["details"]["compatibility_issues"]


def test_crlf_skill_frontmatter_is_reported_loadable(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    skill_file = source / "workspace" / "skills" / "demo" / "SKILL.md"
    skill_file.write_text(
        "---\r\nname: demo\r\ndescription: Demo skill\r\n---\r\nDemo body\r\n",
        encoding="utf-8",
    )
    home = tmp_path / "opensquilla-home"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    report = OpenClawMigrator(
        MigrationOptions(
            source=source,
            config_path=tmp_path / "config.toml",
            apply=True,
            preset="user-data",
        )
    ).migrate()

    item = next(item for item in report["items"] if item["kind"] == "skills")
    assert item["details"]["opensquilla_loadable"] is True
    assert item["details"]["compatibility"] == "loadable"


def test_bom_encoded_openclaw_files_are_parsed(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    config = json.loads((source / "openclaw.json").read_text(encoding="utf-8"))
    config["agents"]["defaults"]["model"] = "claude-3-5-sonnet-latest"
    config["models"] = {"providers": {"anthropic": {"baseUrl": "https://example.test/v1"}}}
    (source / "openclaw.json").write_text(
        json.dumps(config),
        encoding="utf-8-sig",
    )
    (source / ".env").write_text(
        "ANTHROPIC_API_KEY=sk-bom-secret\n",
        encoding="utf-8-sig",
    )
    (source / "workspace" / "skills" / "demo" / "SKILL.md").write_text(
        "---\nname: demo\ndescription: Demo skill\n---\nDemo body\n",
        encoding="utf-8-sig",
    )
    home = tmp_path / "opensquilla-home"
    config_path = tmp_path / "config.toml"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    report = OpenClawMigrator(
        MigrationOptions(
            source=source,
            config_path=config_path,
            apply=True,
            migrate_secrets=True,
        )
    ).migrate()

    config = tomllib.loads(config_path.read_text(encoding="utf-8"))
    assert config["llm"]["provider"] == "anthropic"
    assert config["llm"]["api_key_env"] == "ANTHROPIC_API_KEY"
    assert config["llm"]["base_url"] == "https://example.test/v1"
    assert "ANTHROPIC_API_KEY=sk-bom-secret" in (home / ".env").read_text(
        encoding="utf-8"
    )
    item = next(item for item in report["items"] if item["kind"] == "skills")
    assert item["details"]["opensquilla_loadable"] is True


def test_model_config_resolves_object_and_alias_catalog(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    config = json.loads((source / "openclaw.json").read_text(encoding="utf-8"))
    config["agents"]["defaults"]["model"] = {"primary": "Claude Opus 4.6"}
    config["agents"]["defaults"]["models"] = {
        "anthropic/claude-opus-4-6": {"alias": "Claude Opus 4.6"}
    }
    (source / "openclaw.json").write_text(json.dumps(config), encoding="utf-8")
    home = tmp_path / "opensquilla-home"
    config_path = tmp_path / "config.toml"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    report = OpenClawMigrator(
        MigrationOptions(source=source, config_path=config_path, apply=True)
    ).migrate()

    persisted = tomllib.loads(config_path.read_text(encoding="utf-8"))
    assert persisted["llm"]["model"] == "anthropic/claude-opus-4-6"
    item = next(item for item in report["items"] if item["kind"] == "model-config")
    assert item["details"]["requested_model"] == "Claude Opus 4.6"
    assert item["details"]["resolved_from_alias"] is True
    assert item["details"]["source_format"] == "object"


def test_provider_keys_can_migrate_from_provider_config_when_opted_in(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    config = json.loads((source / "openclaw.json").read_text(encoding="utf-8"))
    config["agents"]["defaults"]["model"] = "openrouter/deepseek/deepseek-v3.1"
    config["models"] = {
        "providers": {
            "openrouter": {
                "apiKey": "sk-or-from-config",
                "baseUrl": "https://openrouter.example.test/api/v1",
            }
        }
    }
    (source / "openclaw.json").write_text(json.dumps(config), encoding="utf-8")
    home = tmp_path / "opensquilla-home"
    config_path = tmp_path / "config.toml"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    no_secret_report = OpenClawMigrator(
        MigrationOptions(source=source, config_path=tmp_path / "no-secret.toml", apply=True)
    ).migrate()
    no_secret_env = (home / ".env").read_text(encoding="utf-8") if (home / ".env").exists() else ""
    assert "OPENROUTER_API_KEY" not in no_secret_env
    assert "sk-or-from-config" not in json.dumps(no_secret_report)

    secret_report = OpenClawMigrator(
        MigrationOptions(
            source=source,
            config_path=config_path,
            apply=True,
            migrate_secrets=True,
        )
    ).migrate()

    env_text = (home / ".env").read_text(encoding="utf-8")
    assert "OPENROUTER_API_KEY=sk-or-from-config" in env_text
    persisted = tomllib.loads(config_path.read_text(encoding="utf-8"))
    # Sparse persistence omits values equal to the built-in default
    # (provider "openrouter"); the effective provider must still resolve
    # to the migrated value.
    assert persisted["llm"].get("provider", "openrouter") == "openrouter"
    assert load_config(config_path).llm.provider == "openrouter"
    assert persisted["llm"]["model"] == "deepseek/deepseek-v3.1"
    assert persisted["llm"]["api_key_env"] == "OPENROUTER_API_KEY"
    assert persisted["llm"]["base_url"] == "https://openrouter.example.test/api/v1"
    item = next(item for item in secret_report["items"] if item["kind"] == "model-config")
    assert item["details"]["source_model"] == "openrouter/deepseek/deepseek-v3.1"
    assert item["details"]["normalized_provider_prefix"] == "openrouter"
    assert "sk-or-from-config" not in json.dumps(secret_report)


def test_zai_and_glm_models_migrate_to_zhipu_provider(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    config = json.loads((source / "openclaw.json").read_text(encoding="utf-8"))
    config["agents"]["defaults"]["model"] = "zai/glm-4.5"
    config["models"] = {
        "providers": {
            "zai": {
                "apiKey": "sk-zai-secret",
                "baseUrl": "https://zhipu.example.test/api/paas/v4",
            }
        }
    }
    (source / "openclaw.json").write_text(json.dumps(config), encoding="utf-8")
    home = tmp_path / "opensquilla-home"
    config_path = tmp_path / "config.toml"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    OpenClawMigrator(
        MigrationOptions(
            source=source,
            config_path=config_path,
            apply=True,
            migrate_secrets=True,
        )
    ).migrate()

    persisted = tomllib.loads(config_path.read_text(encoding="utf-8"))
    assert persisted["llm"]["provider"] == "zhipu"
    assert persisted["llm"]["model"] == "glm-4.5"
    assert persisted["llm"]["api_key_env"] == "ZAI_API_KEY"
    assert persisted["llm"]["base_url"] == "https://zhipu.example.test/api/paas/v4"
    build_provider(persisted["llm"]["provider"], persisted["llm"]["model"])


def test_model_provider_conflict_with_existing_tier_profile_is_reported(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    config = json.loads((source / "openclaw.json").read_text(encoding="utf-8"))
    config["agents"]["defaults"]["model"] = "anthropic/claude-3-5-sonnet"
    (source / "openclaw.json").write_text(json.dumps(config), encoding="utf-8")
    home = tmp_path / "opensquilla-home"
    config_path = tmp_path / "config.toml"
    config_path.write_text(
        "\n".join(
            [
                "[llm]",
                'provider = "openrouter"',
                'model = "deepseek/deepseek-v4-flash"',
                "",
                "[squilla_router]",
                "enabled = true",
                'tier_profile = "openrouter"',
                "",
            ]
        ),
        encoding="utf-8",
    )
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    report = OpenClawMigrator(
        MigrationOptions(source=source, config_path=config_path, apply=True)
    ).migrate()

    persisted = tomllib.loads(config_path.read_text(encoding="utf-8"))
    assert persisted["llm"]["provider"] == "openrouter"
    assert persisted["llm"]["model"] == "deepseek/deepseek-v4-flash"
    assert persisted["squilla_router"]["tier_profile"] == "openrouter"
    item = next(item for item in report["items"] if item["kind"] == "model-config")
    assert item["details"]["tier_profile_conflict"] == "openrouter"
    assert item["details"]["llm_provider_left_unchanged"] == "openrouter"
    assert item["details"]["llm_model_left_unchanged"] == "deepseek/deepseek-v4-flash"
    assert item["details"]["skipped_model"] == "anthropic/claude-3-5-sonnet"
    assert item["details"]["manual_steps"]


def test_model_provider_conflict_with_direct_provider_preserves_model(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    config = json.loads((source / "openclaw.json").read_text(encoding="utf-8"))
    config["agents"]["defaults"]["model"] = "anthropic/claude-3-5-sonnet"
    (source / "openclaw.json").write_text(json.dumps(config), encoding="utf-8")
    home = tmp_path / "opensquilla-home"
    config_path = tmp_path / "config.toml"
    config_path.write_text(
        "\n".join(
            [
                "[llm]",
                'provider = "deepseek"',
                'model = "deepseek-v4-flash"',
                "",
                "[squilla_router]",
                "enabled = true",
                'tier_profile = "deepseek"',
                "",
            ]
        ),
        encoding="utf-8",
    )
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    report = OpenClawMigrator(
        MigrationOptions(source=source, config_path=config_path, apply=True)
    ).migrate()

    persisted = tomllib.loads(config_path.read_text(encoding="utf-8"))
    assert persisted["llm"]["provider"] == "deepseek"
    assert persisted["llm"]["model"] == "deepseek-v4-flash"
    item = next(item for item in report["items"] if item["kind"] == "model-config")
    assert item["details"]["tier_profile_conflict"] == "deepseek"
    assert item["details"]["llm_model_left_unchanged"] == "deepseek-v4-flash"
    assert item["details"]["skipped_model"] == "anthropic/claude-3-5-sonnet"


def test_env_secret_migration_preserves_existing_lines_and_dedupes_keys(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    config = json.loads((source / "openclaw.json").read_text(encoding="utf-8"))
    config["agents"]["defaults"]["model"] = "openrouter/deepseek/deepseek-v3.1"
    (source / "openclaw.json").write_text(json.dumps(config), encoding="utf-8")
    (source / ".env").write_text("OPENROUTER_API_KEY=sk-new\n", encoding="utf-8")
    home = tmp_path / "opensquilla-home"
    env_path = home / ".env"
    env_path.parent.mkdir(parents=True)
    env_path.write_text(
        "\n".join(
            [
                "# existing operator note",
                "",
                "UNRELATED=value",
                "export OPENROUTER_API_KEY=sk-old",
                "",
            ]
        ),
        encoding="utf-8",
    )
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    OpenClawMigrator(
        MigrationOptions(
            source=source,
            config_path=tmp_path / "config.toml",
            apply=True,
            migrate_secrets=True,
        )
    ).migrate()

    env_text = env_path.read_text(encoding="utf-8")
    assert "# existing operator note" in env_text
    assert "UNRELATED=value" in env_text
    assert "OPENROUTER_API_KEY=sk-new" in env_text
    assert "sk-old" not in env_text
    assert env_text.count("OPENROUTER_API_KEY=") == 1


def test_memory_migration_deduplicates_and_archives_overflow(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    workspace = source / "workspace"
    (workspace / "MEMORY.md").write_text("shared memory\n", encoding="utf-8")
    (workspace / "memory" / "2026-05-10.md").write_text("shared memory\n", encoding="utf-8")
    (workspace / "memory" / "2026-05-11.md").write_text("x" * 90_000, encoding="utf-8")
    home = tmp_path / "opensquilla-home"
    config_path = tmp_path / "config.toml"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    report = OpenClawMigrator(
        MigrationOptions(source=source, config_path=config_path, apply=True)
    ).migrate()

    migrated = (home / "workspace" / "MEMORY.md").read_text(encoding="utf-8")
    assert migrated.count("shared memory") == 1
    assert "Migration overflow" in migrated
    overflow = Path(report["output_dir"]) / "archive" / "memory-overflow" / "MEMORY.overflow.md"
    assert overflow.is_file()
    assert "x" in overflow.read_text(encoding="utf-8")
    item = next(item for item in report["items"] if item["kind"] == "memory")
    assert item["details"]["deduplicated_blocks"] == 1
    assert "overflow" not in item["details"]


def test_workspace_text_is_rebranded_and_original_is_archived(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    workspace = source / "workspace"
    (workspace / "SOUL.md").write_text(
        "OpenClaw should read .openclaw paths. ClawdBot helps MoltBot.\n",
        encoding="utf-8",
    )
    home = tmp_path / "opensquilla-home"
    config_path = tmp_path / "config.toml"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    report = OpenClawMigrator(
        MigrationOptions(source=source, config_path=config_path, apply=True)
    ).migrate()

    migrated = (home / "workspace" / "SOUL.md").read_text(encoding="utf-8")
    assert "OpenSquilla should read .opensquilla paths" in migrated
    assert "ClawdBot" not in migrated
    original = Path(report["output_dir"]) / "archive" / "files" / "workspace-original" / "SOUL.md"
    assert original.read_text(encoding="utf-8").startswith("OpenClaw should read .openclaw")
    item = next(item for item in report["items"] if item["kind"] == "soul")
    assert item["details"]["semantic_conversions"] == ["openclaw-branding"]


def test_rebrand_preserves_openclaw_source_references(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    workspace = source / "workspace"
    (workspace / "SOUL.md").write_text(
        "\n".join(
            [
                "你是 OpenClaw 迁移后的助手。",
                "保留 OpenClaw 来源归档用于人工复核。",
                "不要只迁移 openclaw.json。",
                "今天通过真实 OpenClaw CLI onboarding 和 gateway 验证流程。",
                "OpenClaw 的 branding 文本需要语义转换。",
            ]
        ),
        encoding="utf-8",
    )
    home = tmp_path / "opensquilla-home"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    OpenClawMigrator(
        MigrationOptions(source=source, config_path=tmp_path / "config.toml", apply=True)
    ).migrate()

    migrated = (home / "workspace" / "SOUL.md").read_text(encoding="utf-8")
    assert "你是 OpenSquilla 迁移后的助手。" in migrated
    assert "OpenClaw 来源归档" in migrated
    assert "openclaw.json" in migrated
    assert "OpenClaw CLI onboarding" in migrated
    assert "OpenClaw 的 branding 文本" in migrated


def test_agent_search_and_channel_semantics_are_mapped(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    config = json.loads((source / "openclaw.json").read_text(encoding="utf-8"))
    config["agents"]["defaults"].update(
        {
            "thinkingDefault": "high",
            "compaction": {"mode": "truncate", "model": "legacy-compact-model"},
            "verboseDefault": True,
            "humanDelay": {"minMs": 10, "maxMs": 20},
            "userTimezone": "Asia/Shanghai",
        }
    )
    config["messages"]["telegram"]["allowFrom"] = ["1001", "1002"]
    config["messages"]["discord"]["allowedUsers"] = ["discord-user"]
    config["messages"]["slack"] = {"adminUsers": ["slack-admin"]}
    (source / "openclaw.json").write_text(json.dumps(config), encoding="utf-8")
    (source / ".env").write_text("BRAVE_API_KEY=brave-secret\n", encoding="utf-8")
    home = tmp_path / "opensquilla-home"
    config_path = tmp_path / "config.toml"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    report = OpenClawMigrator(
        MigrationOptions(
            source=source,
            config_path=config_path,
            apply=True,
            migrate_secrets=True,
        )
    ).migrate()

    persisted = tomllib.loads(config_path.read_text(encoding="utf-8"))
    assert persisted["llm"]["thinking"] == "high"
    assert persisted["context_overflow_policy"] == "hard_truncate"
    assert persisted["search_provider"] == "brave"
    assert persisted["search_api_key_env"] == "BRAVE_API_KEY"
    assert persisted["channel_admin_senders"] == {"slack": ["slack-admin"]}
    assert "BRAVE_API_KEY=brave-secret" in (home / ".env").read_text(encoding="utf-8")
    notes = Path(report["output_dir"]) / "MIGRATION_NOTES.md"
    assert notes.is_file()
    notes_text = notes.read_text(encoding="utf-8")
    assert "verboseDefault" in notes_text
    assert "humanDelay" in notes_text
    assert "legacy-compact-model" in notes_text
    assert "allowFrom" in notes_text
    assert "allowedUsers" in notes_text


def test_dry_run_report_includes_semantic_notes(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    config = json.loads((source / "openclaw.json").read_text(encoding="utf-8"))
    config["messages"]["telegram"]["allowFrom"] = ["1001"]
    (source / "openclaw.json").write_text(json.dumps(config), encoding="utf-8")
    home = tmp_path / "opensquilla-home"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    report = OpenClawMigrator(
        MigrationOptions(source=source, config_path=tmp_path / "config.toml", apply=False)
    ).migrate()

    assert report["apply"] is False
    assert any("allowFrom" in note for note in report["notes"])
    assert not Path(report["output_dir"], "MIGRATION_NOTES.md").exists()


def test_rebranded_daily_memory_original_is_archived(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    workspace = source / "workspace"
    (workspace / "MEMORY.md").write_text("stable memory\n", encoding="utf-8")
    (workspace / "memory" / "2026-05-12.md").write_text(
        "OpenClaw daily memory\n",
        encoding="utf-8",
    )
    home = tmp_path / "opensquilla-home"
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))

    report = OpenClawMigrator(
        MigrationOptions(source=source, config_path=tmp_path / "config.toml", apply=True)
    ).migrate()

    migrated = (home / "workspace" / "MEMORY.md").read_text(encoding="utf-8")
    assert "OpenSquilla daily memory" in migrated
    original = (
        Path(report["output_dir"])
        / "archive"
        / "files"
        / "workspace-original"
        / "memory"
        / "2026-05-12.md"
    )
    assert original.read_text(encoding="utf-8") == "OpenClaw daily memory\n"


def test_migrated_workspace_and_loadable_skills_are_consumed_by_opensquilla(
    tmp_path: Path,
    monkeypatch,
) -> None:
    source = _make_source(tmp_path)
    home = tmp_path / "opensquilla-home"
    isolated_home = tmp_path / "isolated-home"
    isolated_home.mkdir()
    monkeypatch.setenv("OPENSQUILLA_STATE_DIR", str(home))
    monkeypatch.setenv("HOME", str(isolated_home))
    monkeypatch.setenv("USERPROFILE", str(isolated_home))

    OpenClawMigrator(
        MigrationOptions(
            source=source,
            config_path=tmp_path / "config.toml",
            apply=True,
            preset="user-data",
        )
    ).migrate()

    context = load_context_files(str(home / "workspace"))
    assert context.soul == "opensquilla soul\n"
    assert context.agents == "agent rules\n"
    assert context.user == "user profile\n"
    assert "core memory" in (context.memory or "")
    assert "daily note" in (context.memory or "")

    loader = SkillLoader(
        bundled_dir=None,
        workspace_dir=None,
        managed_dir=None,
        personal_agents_dir=None,
        project_agents_dir=None,
        extra_dirs=[home / "skills" / "openclaw-imports"],
        snapshot_path=tmp_path / "skills_snapshot.json",
    )
    skills = loader.load_all()
    assert [(skill.name, skill.layer) for skill in skills] == [("demo", SkillLayer.EXTRA)]
    assert skills[0].description == "Demo skill"