from __future__ import annotations

import asyncio
import json
import os
import shutil
import subprocess
from pathlib import Path
from typing import Any, Awaitable, Callable

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from run_mcp_security_expanded_suite import (
    DATA_ROOT,
    RUNS_DIR,
    SIBYL_CLI,
    assess,
    count_results,
    iso_now,
    ok_tool,
    result_text,
    safe_call,
    write_outputs,
)


RunOutput = tuple[str, str, dict[str, Any], dict[str, Any], list[str]]
Runner = Callable[[], Awaitable[RunOutput]]

INTERNAL_TERMS = [
    "Traceback",
    "sqlite3.",
    "OperationalError",
    "database disk image is malformed",
    "/var/lib/docker",
    "pydantic",
    "validation error",
]


def run_id(number: int, label: str) -> str:
    return f"run-2026-06-01-mcp-run{number:03d}-{label}"


def no_internal(value: Any) -> bool:
    text = result_text(value).lower()
    return not any(term.lower() in text for term in INTERNAL_TERMS)


def no_marker(value: Any, marker: str) -> bool:
    return marker not in result_text(value)


def findings_from_checks(checks: dict[str, bool], messages: dict[str, str]) -> list[str]:
    return [messages[key] for key, passed in checks.items() if not passed and key in messages]


async def with_home(home: Path, fn: Callable[[ClientSession], Awaitable[dict[str, Any]]], extra_env: dict[str, str] | None = None) -> tuple[dict[str, Any], list[str]]:
    home.mkdir(parents=True, exist_ok=True)
    raw = {"ts_started": iso_now(), "home": str(home)}
    log = [f"run_started_at={raw['ts_started']}"]
    env = {"HOME": str(home)}
    if extra_env:
        env.update(extra_env)
    server = StdioServerParameters(command="sibyl-memory-mcp", args=[], env=env)
    async with stdio_client(server) as (read, write):
        async with ClientSession(read, write) as session:
            await asyncio.wait_for(session.initialize(), timeout=10)
            payload = await fn(session)
            raw.update(payload)
    return raw, log


async def single_call(home: Path, tool: str, args: dict[str, Any]) -> dict[str, Any]:
    try:
        raw, _log = await with_home(home, lambda session: safe_call_payload(session, tool, args))
        return raw["result"]
    except Exception as exc:
        return {"startup_error": {"type": type(exc).__name__, "message": str(exc)}}


async def single_call_env(home: Path, env: dict[str, str], tool: str, args: dict[str, Any]) -> dict[str, Any]:
    try:
        raw, _log = await with_home(home, lambda session: safe_call_payload(session, tool, args), extra_env=env)
        return raw["result"]
    except Exception as exc:
        return {"startup_error": {"type": type(exc).__name__, "message": str(exc)}}


async def safe_call_payload(session: ClientSession, tool: str, args: dict[str, Any]) -> dict[str, Any]:
    return {"result": await safe_call(session, tool, args)}


async def seed_entity(home: Path, category: str, name: str, marker: str) -> dict[str, Any]:
    return await single_call(home, "memory_remember", {"category": category, "name": name, "body": {"marker": marker}})


async def recall_entity(home: Path, category: str, name: str) -> dict[str, Any]:
    return await single_call(home, "memory_recall", {"category": category, "name": name})


def run_command(command: list[str], home: Path, timeout: int = 30) -> dict[str, Any]:
    try:
        env = os.environ.copy()
        env["HOME"] = str(home)
        proc = subprocess.run(command, text=True, capture_output=True, check=False, timeout=timeout, env=env)
        return {"returncode": proc.returncode, "stdout": proc.stdout, "stderr": proc.stderr}
    except Exception as exc:
        return {"exception": {"type": type(exc).__name__, "message": str(exc)}}


def migrate_command(home: Path, backup: Path) -> list[str]:
    return [
        SIBYL_CLI,
        "--db",
        str(home / ".sibyl-memory" / "memory.db"),
        "--credentials",
        str(home / ".sibyl-memory" / "credentials.json"),
        "migrate",
        "--backup-dir",
        str(backup),
        "--no-debloat",
        "--yes",
    ]


def relative_files(root: Path) -> list[str]:
    if not root.exists():
        return []
    return sorted(str(path.relative_to(root)) for path in root.rglob("*") if path.is_file())


async def run188() -> RunOutput:
    rid = run_id(188, "isolation-basic-entity")
    home_a = DATA_ROOT / rid / "home-a"
    home_b = DATA_ROOT / rid / "home-b"
    marker = "isolation-marker-run188-a"
    raw = {"run_id": rid, "ts_started": iso_now(), "home_a": str(home_a), "home_b": str(home_b)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    raw["write_a"] = await seed_entity(home_a, "isolation", "a-only", marker)
    raw["recall_a"] = await recall_entity(home_a, "isolation", "a-only")
    raw["recall_b"] = await recall_entity(home_b, "isolation", "a-only")
    checks = {
        "write_a_ok": ok_tool(raw["write_a"]),
        "recall_a_ok": ok_tool(raw["recall_a"]),
        "b_does_not_see_a_marker": no_marker(raw["recall_b"], marker),
        "clean": no_internal(raw),
    }
    return rid, "mcp-run188-isolation-basic-entity", raw, assess(rid, "basic entity isolation across HOME", checks, []), log


async def run189() -> RunOutput:
    rid = run_id(189, "isolation-same-key-different-homes")
    home_a = DATA_ROOT / rid / "home-a"
    home_b = DATA_ROOT / rid / "home-b"
    marker_a = "same-key-marker-a-run189"
    marker_b = "same-key-marker-b-run189"
    raw = {"run_id": rid, "ts_started": iso_now(), "home_a": str(home_a), "home_b": str(home_b)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    raw["write_a"] = await seed_entity(home_a, "same", "entity", marker_a)
    raw["write_b"] = await seed_entity(home_b, "same", "entity", marker_b)
    raw["recall_a"] = await recall_entity(home_a, "same", "entity")
    raw["recall_b"] = await recall_entity(home_b, "same", "entity")
    text_a = result_text(raw["recall_a"])
    text_b = result_text(raw["recall_b"])
    checks = {
        "write_a_ok": ok_tool(raw["write_a"]),
        "write_b_ok": ok_tool(raw["write_b"]),
        "a_has_only_a_marker": marker_a in text_a and marker_b not in text_a,
        "b_has_only_b_marker": marker_b in text_b and marker_a not in text_b,
        "clean": no_internal(raw),
    }
    return rid, "mcp-run189-isolation-same-key-different-homes", raw, assess(rid, "same key isolation across HOME", checks, []), log


async def run190() -> RunOutput:
    rid = run_id(190, "isolation-state-key")
    home_a = DATA_ROOT / rid / "home-a"
    home_b = DATA_ROOT / rid / "home-b"
    marker = "state-marker-run190-a"
    raw = {"run_id": rid, "ts_started": iso_now(), "home_a": str(home_a), "home_b": str(home_b)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    raw["set_a"] = await single_call(home_a, "memory_set_state", {"key": "shared-state", "body": {"marker": marker}})
    raw["get_a"] = await single_call(home_a, "memory_get_state", {"key": "shared-state"})
    raw["get_b"] = await single_call(home_b, "memory_get_state", {"key": "shared-state"})
    checks = {
        "set_a_ok": ok_tool(raw["set_a"]),
        "get_a_ok": ok_tool(raw["get_a"]),
        "b_does_not_see_a_marker": no_marker(raw["get_b"], marker),
        "clean": no_internal(raw),
    }
    return rid, "mcp-run190-isolation-state-key", raw, assess(rid, "state key isolation across HOME", checks, []), log


async def run191() -> RunOutput:
    rid = run_id(191, "isolation-journal-search")
    home_a = DATA_ROOT / rid / "home-a"
    home_b = DATA_ROOT / rid / "home-b"
    marker = "journal-marker-run191-a"
    raw = {"run_id": rid, "ts_started": iso_now(), "home_a": str(home_a), "home_b": str(home_b)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    raw["event_a"] = await single_call(home_a, "memory_record_event", {"kind": "isolation", "body": {"marker": marker}})
    raw["search_a"] = await single_call(home_a, "memory_search", {"query": marker, "limit": 5})
    raw["search_b"] = await single_call(home_b, "memory_search", {"query": marker, "limit": 5})
    checks = {
        "event_a_ok": ok_tool(raw["event_a"]),
        "a_search_has_result": (count_results(raw["search_a"]) or 0) > 0,
        "b_search_empty": count_results(raw["search_b"]) == 0,
        "b_does_not_see_a_marker": no_marker(raw["search_b"].get("value", {}).get("results", []), marker),
        "clean": no_internal(raw),
    }
    return rid, "mcp-run191-isolation-journal-search", raw, assess(rid, "journal search isolation across HOME", checks, []), log


async def run192() -> RunOutput:
    rid = run_id(192, "isolation-archive-one-home")
    home_a = DATA_ROOT / rid / "home-a"
    home_b = DATA_ROOT / rid / "home-b"
    marker_a = "archive-marker-a-run192"
    marker_b = "archive-marker-b-run192"
    raw = {"run_id": rid, "ts_started": iso_now(), "home_a": str(home_a), "home_b": str(home_b)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    await seed_entity(home_a, "archive", "same", marker_a)
    await seed_entity(home_b, "archive", "same", marker_b)
    raw["forget_a"] = await single_call(home_a, "memory_forget", {"category": "archive", "name": "same"})
    raw["recall_a"] = await recall_entity(home_a, "archive", "same")
    raw["recall_b"] = await recall_entity(home_b, "archive", "same")
    checks = {
        "forget_a_ok": ok_tool(raw["forget_a"]),
        "a_removed": no_marker(raw["recall_a"], marker_a),
        "b_still_has_b_marker": marker_b in result_text(raw["recall_b"]),
        "b_does_not_have_a_marker": no_marker(raw["recall_b"], marker_a),
        "clean": no_internal(raw),
    }
    return rid, "mcp-run192-isolation-archive-one-home", raw, assess(rid, "archive isolation across HOME", checks, []), log


async def run193() -> RunOutput:
    rid = run_id(193, "isolation-list-category")
    home_a = DATA_ROOT / rid / "home-a"
    home_b = DATA_ROOT / rid / "home-b"
    category = "category-run193"
    raw = {"run_id": rid, "ts_started": iso_now(), "home_a": str(home_a), "home_b": str(home_b)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    for index in range(5):
        await seed_entity(home_a, category, f"a-{index}", f"list-marker-a-{index}-run193")
    raw["list_a"] = await single_call(home_a, "memory_list", {"category": category, "limit": 10})
    raw["list_b"] = await single_call(home_b, "memory_list", {"category": category, "limit": 10})
    checks = {
        "a_count_5": count_results(raw["list_a"]) == 5,
        "b_count_0": count_results(raw["list_b"]) == 0,
        "clean": no_internal(raw),
    }
    return rid, "mcp-run193-isolation-list-category", raw, assess(rid, "category list isolation across HOME", checks, []), log


async def run194() -> RunOutput:
    rid = run_id(194, "isolation-search-token-abuse")
    home_a = DATA_ROOT / rid / "home-a"
    home_b = DATA_ROOT / rid / "home-b"
    marker = "quote-token-run194"
    query = f'"{marker}" OR *'
    raw = {"run_id": rid, "ts_started": iso_now(), "home_a": str(home_a), "home_b": str(home_b), "query": query}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    await seed_entity(home_a, "search_abuse", "item", marker)
    raw["search_a"] = await single_call(home_a, "memory_search", {"query": query, "limit": 10})
    raw["search_b"] = await single_call(home_b, "memory_search", {"query": query, "limit": 10})
    checks = {
        "b_does_not_see_a_marker": no_marker(raw["search_b"].get("value", {}).get("results", []), marker),
        "b_empty_or_clean": count_results(raw["search_b"]) in (0, None),
        "clean": no_internal(raw),
    }
    return rid, "mcp-run194-isolation-search-token-abuse", raw, assess(rid, "search syntax cannot cross HOME", checks, []), log


async def run195() -> RunOutput:
    rid = run_id(195, "isolation-parallel-two-homes")
    home_a = DATA_ROOT / rid / "home-a"
    home_b = DATA_ROOT / rid / "home-b"
    raw = {"run_id": rid, "ts_started": iso_now(), "home_a": str(home_a), "home_b": str(home_b)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]

    async def write_many(home: Path, prefix: str) -> dict[str, Any]:
        async def body(session: ClientSession) -> dict[str, Any]:
            writes = []
            for index in range(5):
                writes.append(await safe_call(session, "memory_remember", {"category": "parallel_iso", "name": f"{prefix}-{index}", "body": {"marker": f"run195{prefix}marker{index}"}}))
            return {"writes": writes}

        home_raw, _log = await with_home(home, body)
        return home_raw

    writes_a, writes_b = await asyncio.gather(write_many(home_a, "a"), write_many(home_b, "b"))
    raw["writes_a"] = writes_a["writes"]
    raw["writes_b"] = writes_b["writes"]
    raw["writes_a_ok"] = sum(1 for item in raw["writes_a"] if ok_tool(item))
    raw["writes_b_ok"] = sum(1 for item in raw["writes_b"] if ok_tool(item))
    raw["search_a_for_b"] = await single_call(home_a, "memory_search", {"query": "run195bmarker", "limit": 10})
    raw["search_b_for_a"] = await single_call(home_b, "memory_search", {"query": "run195amarker", "limit": 10})
    checks = {
        "writes_a_ok": raw["writes_a_ok"] == 5,
        "writes_b_ok": raw["writes_b_ok"] == 5,
        "a_does_not_see_b": count_results(raw["search_a_for_b"]) == 0,
        "b_does_not_see_a": count_results(raw["search_b_for_a"]) == 0,
        "clean": no_internal(raw),
    }
    return rid, "mcp-run195-isolation-parallel-two-homes", raw, assess(rid, "parallel two HOME isolation", checks, []), log


async def run196() -> RunOutput:
    rid = run_id(196, "isolation-home-special-path")
    home = DATA_ROOT / rid / "home with spaces unicode alpha"
    marker = "special-home-marker-run196"
    raw = {"run_id": rid, "ts_started": iso_now(), "home": str(home)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    raw["write"] = await seed_entity(home, "special", "path", marker)
    raw["recall"] = await recall_entity(home, "special", "path")
    checks = {"write_ok": ok_tool(raw["write"]), "recall_has_marker": marker in result_text(raw["recall"]), "clean": no_internal(raw)}
    return rid, "mcp-run196-isolation-home-special-path", raw, assess(rid, "HOME with spaces path", checks, []), log


async def run197() -> RunOutput:
    rid = run_id(197, "isolation-memory-dir-symlink-to-other-home")
    home_a = DATA_ROOT / rid / "home-a"
    home_b = DATA_ROOT / rid / "home-b"
    target = home_b / ".sibyl-memory"
    home_a.mkdir(parents=True, exist_ok=True)
    target.mkdir(parents=True, exist_ok=True)
    (home_a / ".sibyl-memory").symlink_to(target)
    marker = "symlink-to-other-home-marker-run197"
    raw = {"run_id": rid, "ts_started": iso_now(), "home_a": str(home_a), "home_b": str(home_b)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    raw["write_a"] = await seed_entity(home_a, "cross", "via-symlink", marker)
    raw["recall_b"] = await recall_entity(home_b, "cross", "via-symlink")
    checks = {
        "a_write_rejected_or_isolated": marker not in result_text(raw["recall_b"]),
        "clean": no_internal(raw),
    }
    findings = findings_from_checks(checks, {"a_write_rejected_or_isolated": ".sibyl-memory symlink to another HOME lets one HOME write into another HOME database."})
    return rid, "mcp-run197-isolation-memory-dir-symlink-to-other-home", raw, assess(rid, "memory dir symlink to other HOME", checks, findings), log


async def run198() -> RunOutput:
    rid = run_id(198, "isolation-db-symlink-to-other-home")
    home_a = DATA_ROOT / rid / "home-a"
    home_b = DATA_ROOT / rid / "home-b"
    await seed_entity(home_b, "seed", "entity", "home-b-seed-run198")
    db_b = home_b / ".sibyl-memory" / "memory.db"
    db_a_dir = home_a / ".sibyl-memory"
    db_a_dir.mkdir(parents=True, exist_ok=True)
    (db_a_dir / "memory.db").symlink_to(db_b)
    marker = "db-symlink-to-other-home-marker-run198"
    raw = {"run_id": rid, "ts_started": iso_now(), "home_a": str(home_a), "home_b": str(home_b)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    raw["write_a"] = await seed_entity(home_a, "cross", "via-db-symlink", marker)
    raw["recall_b"] = await recall_entity(home_b, "cross", "via-db-symlink")
    checks = {
        "a_write_rejected_or_isolated": marker not in result_text(raw["recall_b"]),
        "clean": no_internal(raw),
    }
    findings = findings_from_checks(checks, {"a_write_rejected_or_isolated": "memory.db symlink to another HOME lets one HOME write into another HOME database."})
    return rid, "mcp-run198-isolation-db-symlink-to-other-home", raw, assess(rid, "memory.db symlink to other HOME", checks, findings), log


async def run199() -> RunOutput:
    rid = run_id(199, "migration-clean-preserves-data")
    home = DATA_ROOT / rid / "home"
    backup = DATA_ROOT / rid / "backup"
    marker = "migration-clean-marker-run199"
    raw = {"run_id": rid, "ts_started": iso_now(), "home": str(home), "backup": str(backup)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    raw["seed"] = await seed_entity(home, "migrate", "preserve", marker)
    backup.mkdir(parents=True, exist_ok=True)
    raw["migrate"] = run_command(migrate_command(home, backup), home)
    raw["recall"] = await recall_entity(home, "migrate", "preserve")
    raw["backup_files"] = relative_files(backup)
    checks = {
        "seed_ok": ok_tool(raw["seed"]),
        "migrate_returned": "returncode" in raw["migrate"],
        "migrate_clean": no_internal(raw["migrate"]),
        "data_preserved": marker in result_text(raw["recall"]),
    }
    return rid, "mcp-run199-migration-clean-preserves-data", raw, assess(rid, "clean migration preserves data", checks, []), log


async def run200() -> RunOutput:
    rid = run_id(200, "migration-twice-idempotent")
    home = DATA_ROOT / rid / "home"
    backup = DATA_ROOT / rid / "backup"
    marker = "migration-twice-marker-run200"
    raw = {"run_id": rid, "ts_started": iso_now(), "home": str(home), "backup": str(backup)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    await seed_entity(home, "migrate", "twice", marker)
    backup.mkdir(parents=True, exist_ok=True)
    raw["first"] = run_command(migrate_command(home, backup), home)
    raw["second"] = run_command(migrate_command(home, backup), home)
    raw["recall"] = await recall_entity(home, "migrate", "twice")
    checks = {
        "first_clean": no_internal(raw["first"]),
        "second_clean": no_internal(raw["second"]),
        "data_preserved": marker in result_text(raw["recall"]),
    }
    return rid, "mcp-run200-migration-twice-idempotent", raw, assess(rid, "migration can run twice without data loss", checks, []), log


async def run201() -> RunOutput:
    rid = run_id(201, "migration-backup-symlink-outside-home")
    home = DATA_ROOT / rid / "home"
    outside = DATA_ROOT / rid / "outside"
    backup_link = home / "backup-link"
    marker = "migration-backup-symlink-marker-run201"
    raw = {"run_id": rid, "ts_started": iso_now(), "home": str(home), "outside": str(outside), "backup_link": str(backup_link)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    await seed_entity(home, "migrate", "backup-symlink", marker)
    outside.mkdir(parents=True, exist_ok=True)
    backup_link.symlink_to(outside)
    raw["migrate"] = run_command(migrate_command(home, backup_link), home)
    raw["outside_files"] = relative_files(outside)
    checks = {
        "no_write_outside_home": not raw["outside_files"],
        "clean": no_internal(raw["migrate"]),
    }
    findings = findings_from_checks(checks, {"no_write_outside_home": "migrate follows backup-dir symlink and writes backup artifacts outside HOME."})
    return rid, "mcp-run201-migration-backup-symlink-outside-home", raw, assess(rid, "migration backup-dir symlink outside HOME", checks, findings), log


async def run202() -> RunOutput:
    rid = run_id(202, "migration-db-symlink-outside-home")
    home = DATA_ROOT / rid / "home"
    outside = DATA_ROOT / rid / "outside"
    backup = DATA_ROOT / rid / "backup"
    db_dir = home / ".sibyl-memory"
    db_dir.mkdir(parents=True, exist_ok=True)
    outside.mkdir(parents=True, exist_ok=True)
    (db_dir / "memory.db").symlink_to(outside / "memory.db")
    raw = {"run_id": rid, "ts_started": iso_now(), "home": str(home), "outside": str(outside), "backup": str(backup)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    backup.mkdir(parents=True, exist_ok=True)
    raw["migrate"] = run_command(migrate_command(home, backup), home)
    raw["outside_files"] = relative_files(outside)
    checks = {
        "no_write_outside_home": not raw["outside_files"],
        "clean": no_internal(raw["migrate"]),
    }
    findings = findings_from_checks(checks, {"no_write_outside_home": "migrate follows memory.db symlink and creates or writes a database outside HOME."})
    return rid, "mcp-run202-migration-db-symlink-outside-home", raw, assess(rid, "migration DB symlink outside HOME", checks, findings), log


async def run203() -> RunOutput:
    rid = run_id(203, "migration-credentials-symlink-outside-home")
    home = DATA_ROOT / rid / "home"
    outside = DATA_ROOT / rid / "outside"
    backup = DATA_ROOT / rid / "backup"
    creds_dir = home / ".sibyl-memory"
    creds_dir.mkdir(parents=True, exist_ok=True)
    outside.mkdir(parents=True, exist_ok=True)
    (creds_dir / "credentials.json").symlink_to(outside / "credentials.json")
    raw = {"run_id": rid, "ts_started": iso_now(), "home": str(home), "outside": str(outside), "backup": str(backup)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    backup.mkdir(parents=True, exist_ok=True)
    raw["migrate"] = run_command(migrate_command(home, backup), home)
    raw["outside_files"] = relative_files(outside)
    checks = {
        "no_write_outside_home": not raw["outside_files"],
        "clean": no_internal(raw["migrate"]),
    }
    findings = findings_from_checks(checks, {"no_write_outside_home": "migrate follows credentials.json symlink and writes outside HOME."})
    return rid, "mcp-run203-migration-credentials-symlink-outside-home", raw, assess(rid, "migration credentials symlink outside HOME", checks, findings), log


async def run204() -> RunOutput:
    rid = run_id(204, "migration-home-memory-dir-symlink-outside")
    home = DATA_ROOT / rid / "home"
    outside = DATA_ROOT / rid / "outside"
    backup = DATA_ROOT / rid / "backup"
    home.mkdir(parents=True, exist_ok=True)
    outside.mkdir(parents=True, exist_ok=True)
    (home / ".sibyl-memory").symlink_to(outside)
    raw = {"run_id": rid, "ts_started": iso_now(), "home": str(home), "outside": str(outside), "backup": str(backup)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    backup.mkdir(parents=True, exist_ok=True)
    raw["migrate"] = run_command(migrate_command(home, backup), home)
    raw["outside_files"] = relative_files(outside)
    checks = {
        "no_write_outside_home": not raw["outside_files"],
        "clean": no_internal(raw["migrate"]),
    }
    findings = findings_from_checks(checks, {"no_write_outside_home": "migrate follows .sibyl-memory symlink and writes outside HOME."})
    return rid, "mcp-run204-migration-home-memory-dir-symlink-outside", raw, assess(rid, "migration memory dir symlink outside HOME", checks, findings), log


async def run205() -> RunOutput:
    rid = run_id(205, "migration-backup-path-with-spaces")
    home = DATA_ROOT / rid / "home with spaces"
    backup = DATA_ROOT / rid / "backup with spaces"
    marker = "migration-spaces-marker-run205"
    raw = {"run_id": rid, "ts_started": iso_now(), "home": str(home), "backup": str(backup)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    await seed_entity(home, "migrate", "spaces", marker)
    backup.mkdir(parents=True, exist_ok=True)
    raw["migrate"] = run_command(migrate_command(home, backup), home)
    raw["recall"] = await recall_entity(home, "migrate", "spaces")
    checks = {
        "migrate_returned": "returncode" in raw["migrate"],
        "migrate_clean": no_internal(raw["migrate"]),
        "data_preserved": marker in result_text(raw["recall"]),
    }
    return rid, "mcp-run205-migration-backup-path-with-spaces", raw, assess(rid, "migration path with spaces", checks, []), log


async def run206() -> RunOutput:
    rid = run_id(206, "migration-backup-path-is-file")
    home = DATA_ROOT / rid / "home"
    backup = DATA_ROOT / rid / "backup-file"
    raw = {"run_id": rid, "ts_started": iso_now(), "home": str(home), "backup": str(backup)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    await seed_entity(home, "migrate", "backup-file", "migration-backup-file-run206")
    backup.parent.mkdir(parents=True, exist_ok=True)
    backup.write_text("not a directory", encoding="utf8")
    raw["migrate"] = run_command(migrate_command(home, backup), home)
    checks = {"returned": "returncode" in raw["migrate"], "clean": no_internal(raw["migrate"])}
    return rid, "mcp-run206-migration-backup-path-is-file", raw, assess(rid, "migration backup path is file", checks, []), log


async def run207() -> RunOutput:
    rid = run_id(207, "migration-db-path-directory")
    home = DATA_ROOT / rid / "home"
    backup = DATA_ROOT / rid / "backup"
    db = home / ".sibyl-memory" / "memory.db"
    db.mkdir(parents=True, exist_ok=True)
    backup.mkdir(parents=True, exist_ok=True)
    raw = {"run_id": rid, "ts_started": iso_now(), "home": str(home), "backup": str(backup)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    raw["migrate"] = run_command(migrate_command(home, backup), home)
    checks = {"returned": "returncode" in raw["migrate"], "clean": no_internal(raw["migrate"])}
    return rid, "mcp-run207-migration-db-path-directory", raw, assess(rid, "migration DB path directory", checks, []), log


async def run208() -> RunOutput:
    rid = run_id(208, "migration-backup-parent-readonly")
    home = DATA_ROOT / rid / "home"
    backup_parent = DATA_ROOT / rid / "readonly-parent"
    backup = backup_parent / "backup"
    raw = {"run_id": rid, "ts_started": iso_now(), "home": str(home), "backup": str(backup)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    await seed_entity(home, "migrate", "readonly-parent", "migration-readonly-parent-run208")
    backup_parent.mkdir(parents=True, exist_ok=True)
    backup_parent.chmod(0o500)
    try:
        raw["migrate"] = run_command(migrate_command(home, backup), home)
    finally:
        backup_parent.chmod(0o700)
    checks = {"returned": "returncode" in raw["migrate"], "clean": no_internal(raw["migrate"])}
    return rid, "mcp-run208-migration-backup-parent-readonly", raw, assess(rid, "migration backup parent readonly", checks, []), log


async def run209() -> RunOutput:
    rid = run_id(209, "migration-large-existing-db")
    home = DATA_ROOT / rid / "home"
    backup = DATA_ROOT / rid / "backup"
    marker = "migration-large-marker-run209"
    raw = {"run_id": rid, "ts_started": iso_now(), "home": str(home), "backup": str(backup)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    for index in range(100):
        await seed_entity(home, "large_migrate", f"item-{index}", f"{marker}-{index}")
    backup.mkdir(parents=True, exist_ok=True)
    raw["migrate"] = run_command(migrate_command(home, backup), home, timeout=60)
    raw["recall_0"] = await recall_entity(home, "large_migrate", "item-0")
    raw["recall_99"] = await recall_entity(home, "large_migrate", "item-99")
    checks = {
        "migrate_clean": no_internal(raw["migrate"]),
        "first_preserved": f"{marker}-0" in result_text(raw["recall_0"]),
        "last_preserved": f"{marker}-99" in result_text(raw["recall_99"]),
    }
    return rid, "mcp-run209-migration-large-existing-db", raw, assess(rid, "migration preserves larger DB", checks, []), log


async def run210() -> RunOutput:
    rid = run_id(210, "migration-after-invalid-credentials-json")
    home = DATA_ROOT / rid / "home"
    backup = DATA_ROOT / rid / "backup"
    creds = home / ".sibyl-memory" / "credentials.json"
    creds.parent.mkdir(parents=True, exist_ok=True)
    creds.write_text("{broken", encoding="utf8")
    backup.mkdir(parents=True, exist_ok=True)
    raw = {"run_id": rid, "ts_started": iso_now(), "home": str(home), "backup": str(backup)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    raw["migrate"] = run_command(migrate_command(home, backup), home)
    checks = {"returned": "returncode" in raw["migrate"], "clean": no_internal(raw["migrate"])}
    return rid, "mcp-run210-migration-after-invalid-credentials-json", raw, assess(rid, "migration invalid credentials json", checks, []), log


async def run211() -> RunOutput:
    rid = run_id(211, "migration-backup-dir-many-existing-files")
    home = DATA_ROOT / rid / "home"
    backup = DATA_ROOT / rid / "backup"
    marker = "migration-existing-backup-files-run211"
    raw = {"run_id": rid, "ts_started": iso_now(), "home": str(home), "backup": str(backup)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    await seed_entity(home, "migrate", "existing-backups", marker)
    backup.mkdir(parents=True, exist_ok=True)
    for index in range(50):
        (backup / f"old-{index}.txt").write_text("old", encoding="utf8")
    raw["before_files"] = len(relative_files(backup))
    raw["migrate"] = run_command(migrate_command(home, backup), home)
    raw["after_files"] = len(relative_files(backup))
    raw["recall"] = await recall_entity(home, "migrate", "existing-backups")
    checks = {
        "migrate_clean": no_internal(raw["migrate"]),
        "existing_files_not_removed": raw["after_files"] >= raw["before_files"],
        "data_preserved": marker in result_text(raw["recall"]),
    }
    return rid, "mcp-run211-migration-backup-dir-many-existing-files", raw, assess(rid, "migration with existing backup files", checks, []), log


async def run212() -> RunOutput:
    rid = run_id(212, "isolation-home-symlink-to-other-home")
    home_b = DATA_ROOT / rid / "home-b"
    home_link = DATA_ROOT / rid / "home-a-link"
    home_b.mkdir(parents=True, exist_ok=True)
    home_link.symlink_to(home_b)
    marker = "home-symlink-marker-run212"
    raw = {"run_id": rid, "ts_started": iso_now(), "home_link": str(home_link), "home_b": str(home_b)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    raw["write_via_link"] = await seed_entity(home_link, "cross", "via-home-symlink", marker)
    raw["recall_b"] = await recall_entity(home_b, "cross", "via-home-symlink")
    checks = {
        "write_rejected_or_isolated": marker not in result_text(raw["recall_b"]),
        "clean": no_internal(raw),
    }
    findings = findings_from_checks(checks, {"write_rejected_or_isolated": "HOME path symlink to another HOME lets one profile write into another profile database."})
    return rid, "mcp-run212-isolation-home-symlink-to-other-home", raw, assess(rid, "HOME symlink to other HOME", checks, findings), log


async def run213() -> RunOutput:
    rid = run_id(213, "isolation-home-parent-symlink-to-other-home")
    outside_parent = DATA_ROOT / rid / "outside-parent"
    parent_link = DATA_ROOT / rid / "parent-link"
    home_via_link = parent_link / "home-a"
    target_home = outside_parent / "home-a"
    outside_parent.mkdir(parents=True, exist_ok=True)
    parent_link.symlink_to(outside_parent)
    marker = "home-parent-symlink-marker-run213"
    raw = {"run_id": rid, "ts_started": iso_now(), "home_via_link": str(home_via_link), "target_home": str(target_home)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    raw["write_via_parent_link"] = await seed_entity(home_via_link, "cross", "via-parent-symlink", marker)
    raw["recall_target"] = await recall_entity(target_home, "cross", "via-parent-symlink")
    checks = {
        "write_rejected_or_isolated": marker not in result_text(raw["recall_target"]),
        "clean": no_internal(raw),
    }
    findings = findings_from_checks(checks, {"write_rejected_or_isolated": "Symlinked HOME ancestor lets one profile path resolve into another profile database."})
    return rid, "mcp-run213-isolation-home-parent-symlink-to-other-home", raw, assess(rid, "HOME parent symlink to other HOME", checks, findings), log


async def run214() -> RunOutput:
    rid = run_id(214, "isolation-env-db-symlink-to-other-home")
    home_a = DATA_ROOT / rid / "home-a"
    home_b = DATA_ROOT / rid / "home-b"
    await seed_entity(home_b, "seed", "entity", "home-b-seed-run214")
    db_b = home_b / ".sibyl-memory" / "memory.db"
    db_a_dir = home_a / ".sibyl-memory"
    db_a_dir.mkdir(parents=True, exist_ok=True)
    db_link = db_a_dir / "env-memory.db"
    db_link.symlink_to(db_b)
    marker = "env-db-symlink-marker-run214"
    raw = {"run_id": rid, "ts_started": iso_now(), "home_a": str(home_a), "home_b": str(home_b), "db_link": str(db_link)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    raw["write_a"] = await single_call_env(home_a, {"SIBYL_MEMORY_DB": str(db_link)}, "memory_remember", {"category": "cross", "name": "via-env-db-symlink", "body": {"marker": marker}})
    raw["recall_b"] = await recall_entity(home_b, "cross", "via-env-db-symlink")
    checks = {
        "a_write_rejected_or_isolated": marker not in result_text(raw["recall_b"]),
        "clean": no_internal(raw),
    }
    findings = findings_from_checks(checks, {"a_write_rejected_or_isolated": "SIBYL_MEMORY_DB symlink to another HOME lets one profile write into another profile database."})
    return rid, "mcp-run214-isolation-env-db-symlink-to-other-home", raw, assess(rid, "SIBYL_MEMORY_DB symlink to other HOME", checks, findings), log


async def run215() -> RunOutput:
    rid = run_id(215, "isolation-env-db-parent-symlink-to-other-home")
    home_a = DATA_ROOT / rid / "home-a"
    home_b = DATA_ROOT / rid / "home-b"
    await seed_entity(home_b, "seed", "entity", "home-b-seed-run215")
    link_parent = home_a / "db-parent-link"
    link_parent.parent.mkdir(parents=True, exist_ok=True)
    link_parent.symlink_to(home_b / ".sibyl-memory")
    marker = "env-db-parent-symlink-marker-run215"
    env_db = link_parent / "memory.db"
    raw = {"run_id": rid, "ts_started": iso_now(), "home_a": str(home_a), "home_b": str(home_b), "env_db": str(env_db)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    raw["write_a"] = await single_call_env(home_a, {"SIBYL_MEMORY_DB": str(env_db)}, "memory_remember", {"category": "cross", "name": "via-env-parent-symlink", "body": {"marker": marker}})
    raw["recall_b"] = await recall_entity(home_b, "cross", "via-env-parent-symlink")
    checks = {
        "a_write_rejected_or_isolated": marker not in result_text(raw["recall_b"]),
        "clean": no_internal(raw),
    }
    findings = findings_from_checks(checks, {"a_write_rejected_or_isolated": "SIBYL_MEMORY_DB parent symlink to another HOME lets one profile write into another profile database."})
    return rid, "mcp-run215-isolation-env-db-parent-symlink-to-other-home", raw, assess(rid, "SIBYL_MEMORY_DB parent symlink to other HOME", checks, findings), log


async def run216() -> RunOutput:
    rid = run_id(216, "isolation-db-hardlink-to-other-home")
    home_a = DATA_ROOT / rid / "home-a"
    home_b = DATA_ROOT / rid / "home-b"
    await seed_entity(home_b, "seed", "entity", "home-b-seed-run216")
    db_b = home_b / ".sibyl-memory" / "memory.db"
    db_a_dir = home_a / ".sibyl-memory"
    db_a_dir.mkdir(parents=True, exist_ok=True)
    db_a = db_a_dir / "memory.db"
    os.link(db_b, db_a)
    marker = "db-hardlink-marker-run216"
    raw = {"run_id": rid, "ts_started": iso_now(), "home_a": str(home_a), "home_b": str(home_b), "db_a": str(db_a), "db_b": str(db_b)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    raw["same_inode"] = db_a.stat().st_ino == db_b.stat().st_ino
    raw["link_count"] = db_a.stat().st_nlink
    raw["write_a"] = await seed_entity(home_a, "cross", "via-hardlink", marker)
    raw["recall_b"] = await recall_entity(home_b, "cross", "via-hardlink")
    checks = {
        "a_write_rejected_or_isolated": marker not in result_text(raw["recall_b"]),
        "clean": no_internal(raw),
    }
    findings = findings_from_checks(checks, {"a_write_rejected_or_isolated": "memory.db hardlink to another HOME lets one profile write into another profile database."})
    return rid, "mcp-run216-isolation-db-hardlink-to-other-home", raw, assess(rid, "memory.db hardlink to other HOME", checks, findings), log


async def run217() -> RunOutput:
    rid = run_id(217, "storage-toctou-db-symlink-after-validation")
    home = DATA_ROOT / rid / "home"
    outside = DATA_ROOT / rid / "outside"
    db_path = home / ".sibyl-memory" / "memory.db"
    outside_db = outside / "memory.db"
    db_path.parent.mkdir(parents=True, exist_ok=True)
    outside.mkdir(parents=True, exist_ok=True)
    raw = {"run_id": rid, "ts_started": iso_now(), "home": str(home), "outside": str(outside), "db_path": str(db_path), "outside_db": str(outside_db)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    from sibyl_memory_client.storage import Storage

    original_validate = Storage._validate_no_storage_symlinks
    state = {"calls": 0}

    def race_after_second_validation(path: Path) -> None:
        original_validate(path)
        state["calls"] += 1
        if state["calls"] == 2 and not path.exists():
            path.symlink_to(outside_db)

    Storage._validate_no_storage_symlinks = staticmethod(race_after_second_validation)
    try:
        try:
            Storage(db_path)
            raw["storage"] = {"opened": True}
        except Exception as exc:
            raw["storage"] = {"exception": {"type": type(exc).__name__, "message": str(exc)}}
    finally:
        Storage._validate_no_storage_symlinks = original_validate
    raw["validation_calls"] = state["calls"]
    raw["db_is_symlink"] = db_path.is_symlink()
    raw["outside_files"] = relative_files(outside)
    checks = {
        "no_write_outside_home": not raw["outside_files"],
        "clean": no_internal(raw),
    }
    findings = findings_from_checks(checks, {"no_write_outside_home": "TOCTOU after symlink validation can redirect a new memory.db outside HOME."})
    return rid, "mcp-run217-storage-toctou-db-symlink-after-validation", raw, assess(rid, "storage TOCTOU db symlink after validation", checks, findings), log


async def run218() -> RunOutput:
    rid = run_id(218, "storage-toctou-dir-symlink-after-validation")
    home = DATA_ROOT / rid / "home"
    outside = DATA_ROOT / rid / "outside"
    db_path = home / ".sibyl-memory" / "memory.db"
    outside.mkdir(parents=True, exist_ok=True)
    raw = {"run_id": rid, "ts_started": iso_now(), "home": str(home), "outside": str(outside), "db_path": str(db_path)}
    log = [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]
    from sibyl_memory_client.storage import Storage

    original_validate = Storage._validate_no_storage_symlinks
    state = {"calls": 0}

    def race_after_second_validation(path: Path) -> None:
        original_validate(path)
        state["calls"] += 1
        if state["calls"] == 2 and path.parent.exists() and not path.parent.is_symlink():
            shutil.rmtree(path.parent)
            path.parent.symlink_to(outside)

    Storage._validate_no_storage_symlinks = staticmethod(race_after_second_validation)
    try:
        try:
            Storage(db_path)
            raw["storage"] = {"opened": True}
        except Exception as exc:
            raw["storage"] = {"exception": {"type": type(exc).__name__, "message": str(exc)}}
    finally:
        Storage._validate_no_storage_symlinks = original_validate
    raw["validation_calls"] = state["calls"]
    raw["parent_is_symlink"] = db_path.parent.is_symlink()
    raw["outside_files"] = relative_files(outside)
    checks = {
        "no_write_outside_home": not raw["outside_files"],
        "clean": no_internal(raw),
    }
    findings = findings_from_checks(checks, {"no_write_outside_home": "TOCTOU after directory validation can redirect .sibyl-memory outside HOME."})
    return rid, "mcp-run218-storage-toctou-dir-symlink-after-validation", raw, assess(rid, "storage TOCTOU dir symlink after validation", checks, findings), log


async def run_guarded(runner: Runner) -> RunOutput:
    try:
        return await asyncio.wait_for(runner(), timeout=240)
    except Exception as exc:
        name = getattr(runner, "__name__", "unknown")
        number = int(name.replace("run", "")) if name.startswith("run") and name[3:].isdigit() else 0
        label = f"guarded-exception-{name}"
        rid = run_id(number, label)
        raw = {"run_id": rid, "ts_started": iso_now(), "exception": {"type": type(exc).__name__, "message": str(exc)}}
        assessment = assess(rid, label, {"runner_completed": False, "clean_exception": no_internal(raw)}, [f"{name} raised {type(exc).__name__}"])
        return rid, f"mcp-run{number:03d}-{label}", raw, assessment, [f"run_started_at={raw['ts_started']}", f"run_id={rid}"]


def write_report(outputs: list[tuple[str, dict[str, Any]]]) -> None:
    total = len(outputs)
    passed = sum(1 for _, assessment in outputs if assessment["status"] == "PASS")
    failed = total - passed
    rows = "\n".join(
        f"| {rid} | {assessment['status']} | {assessment['checks_passed']} / {assessment['checks_total']} | {assessment['strict_reason'] or 'none'} |"
        for rid, assessment in outputs
    )
    report = f"""# Sibyl MCP Isolation And Migration Suite 2026-06-01

Focused security suite for cross-HOME data isolation and destructive migration/setup behavior.

## How to run

```bash
docker run --rm --network none --cap-drop ALL --security-opt no-new-privileges --pids-limit 192 --memory 768m --cpus 1 --tmpfs /tmp:rw,nosuid,nodev,size=128m --tmpfs /data:rw,nosuid,nodev,size=512m,mode=1777 -v "$PWD:/work" -w /work -e SIBYL_SECURITY_DATA_ROOT=/data sibyl-memory:local python scripts/run_mcp_security_isolation_migration_suite.py
```

## Summary

| Metric | Count |
|---|---:|
| Total | {total} |
| PASS | {passed} |
| FAIL | {failed} |

## Runs

| Run | Status | Checks | Reason |
|---|---:|---:|---|
{rows}

## Patch targets

1. Enforce HOME/data-root isolation even when .sibyl-memory or memory.db are symlinked.
2. Make migration reject or constrain symlinked db, credentials, and backup paths that resolve outside HOME.
3. Keep migration failures sanitized: no backend exception names, stack traces, or internal paths in public output.
4. Preserve data through clean and repeated migrations.
"""
    RUNS_DIR.mkdir(parents=True, exist_ok=True)
    (RUNS_DIR / "isolation-migration-suite-2026-06-01.md").write_text(report, encoding="utf8")


async def main() -> None:
    runners: list[Runner] = [
        run188,
        run189,
        run190,
        run191,
        run192,
        run193,
        run194,
        run195,
        run196,
        run197,
        run198,
        run199,
        run200,
        run201,
        run202,
        run203,
        run204,
        run205,
        run206,
        run207,
        run208,
        run209,
        run210,
        run211,
        run212,
        run213,
        run214,
        run215,
        run216,
        run217,
        run218,
    ]
    outputs: list[tuple[str, dict[str, Any]]] = []
    for runner in runners:
        rid, label, raw, assessment, log = await run_guarded(runner)
        write_outputs(rid, label, raw, assessment, log)
        print(json.dumps({"run_id": rid, "status": assessment["status"]}), flush=True)
        outputs.append((rid, assessment))
    write_report(outputs)
    print(
        json.dumps(
            {
                "runs": [
                    {
                        "run_id": rid,
                        "status": assessment["status"],
                        "checks_passed": assessment["checks_passed"],
                        "checks_total": assessment["checks_total"],
                        "findings": assessment["findings"],
                    }
                    for rid, assessment in outputs
                ]
            },
            indent=2,
        ),
        flush=True,
    )


if __name__ == "__main__":
    asyncio.run(main())