# Copyright (c) 2026, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors
#
# SPDX-License-Identifier: Apache-2.0

"""Tests for KnowledgeStore, EventLogWriter, and bootstrapping."""

import asyncio
import json
import sys
from pathlib import Path
from uuid import uuid4

import pytest

pytest.importorskip("watchdog")

from ag2 import Agent
from ag2.events import ModelMessage, ModelRequest, ModelResponse, TaskCompleted, TextInput, UnknownEvent
from ag2.knowledge import (
    DefaultBootstrap,
    DiskKnowledgeStore,
    EventLogWriter,
    KnowledgePlugin,
    LockedKnowledgeStore,
    MemoryKnowledgeStore,
    SqliteKnowledgeStore,
)
from ag2.stream import MemoryStream
from ag2.testing import TestConfig


@pytest.mark.asyncio
class TestMemoryKnowledgeStore:
    async def test_read_write(self) -> None:
        store = MemoryKnowledgeStore()
        await store.write("/test.txt", "hello")
        assert await store.read("/test.txt") == "hello"

    async def test_read_nonexistent(self) -> None:
        store = MemoryKnowledgeStore()
        assert await store.read("/missing.txt") is None

    async def test_list_root(self) -> None:
        store = MemoryKnowledgeStore()
        await store.write("/a.txt", "a")
        await store.write("/b/c.txt", "c")
        entries = await store.list("/")
        assert "a.txt" in entries
        assert "b/" in entries

    async def test_list_subdirectory(self) -> None:
        store = MemoryKnowledgeStore()
        await store.write("/log/stream-1.jsonl", "data1")
        await store.write("/log/stream-2.jsonl", "data2")
        entries = await store.list("/log/")
        assert entries == ["stream-1.jsonl", "stream-2.jsonl"]

    async def test_delete_file(self) -> None:
        store = MemoryKnowledgeStore()
        await store.write("/test.txt", "hello")
        await store.delete("/test.txt")
        assert await store.read("/test.txt") is None

    async def test_delete_directory(self) -> None:
        store = MemoryKnowledgeStore()
        await store.write("/dir/a.txt", "a")
        await store.write("/dir/b.txt", "b")
        await store.delete("/dir")
        assert await store.read("/dir/a.txt") is None
        assert await store.read("/dir/b.txt") is None

    async def test_exists(self) -> None:
        store = MemoryKnowledgeStore()
        await store.write("/test.txt", "hello")
        assert await store.exists("/test.txt") is True
        assert await store.exists("/missing.txt") is False

    async def test_exists_directory(self) -> None:
        store = MemoryKnowledgeStore()
        await store.write("/dir/file.txt", "data")
        assert await store.exists("/dir") is True

    async def test_path_normalization(self) -> None:
        store = MemoryKnowledgeStore()
        await store.write("no_leading_slash.txt", "data")
        assert await store.read("/no_leading_slash.txt") == "data"

    async def test_list_empty(self) -> None:
        store = MemoryKnowledgeStore()
        entries = await store.list("/")
        assert entries == []

    @pytest.mark.parametrize("resubscribe", [False, True])
    async def test_repeated_close_preserves_other_subscription(self, resubscribe: bool) -> None:
        store = MemoryKnowledgeStore()
        received: list[str] = []

        async def callback(path: str) -> None:
            received.append(path)

        old = await store.on_change("/watched", callback)
        if resubscribe:
            await old.close()
        active = await store.on_change("/watched", callback)

        await old.close()
        await old.close()
        await store.write("/watched/file.txt", "first")
        assert received == ["/watched/file.txt"]

        await active.close()
        await store.write("/watched/file.txt", "second")
        assert received == ["/watched/file.txt"]


@pytest.mark.asyncio
class TestEventLogWriter:
    async def test_persist_and_load(self) -> None:
        store = MemoryKnowledgeStore()
        writer = EventLogWriter(store)
        stream_id = uuid4()

        events = [
            ModelRequest([TextInput("hello")]),
            TaskCompleted(
                task_id="t1",
                agent_name="analyzer",
                objective="summarize",
                result="done",
                task_stream=MemoryStream().id,
            ),
        ]
        await writer.persist(stream_id, events)

        assert await writer.load(stream_id) == events

    async def test_persist_dropped_segments(self) -> None:
        store = MemoryKnowledgeStore()
        writer = EventLogWriter(store)
        stream_id = uuid4()

        # Simulate two compaction drops + final persist
        dropped1 = [ModelRequest([TextInput("old-1")])]
        dropped2 = [ModelRequest([TextInput("old-2")])]
        final = [ModelRequest([TextInput("recent")])]

        await writer.persist_dropped(stream_id, dropped1)
        await writer.persist_dropped(stream_id, dropped2)
        await writer.persist(stream_id, final)

        # Load should return all in order: dropped-1, dropped-2, final
        assert await writer.load(stream_id) == [*dropped1, *dropped2, *final]

    async def test_persist_dropped_multiple_writers_no_overwrite(self) -> None:
        """Multiple EventLogWriter instances must not overwrite each other's segments.

        This is the fix for Bug A: segment counter was per-instance, causing
        the second writer to reuse segment number 1 and overwrite the first.
        """
        store = MemoryKnowledgeStore()
        stream_id = uuid4()

        # First compaction — writer 1
        writer1 = EventLogWriter(store)
        await writer1.persist_dropped(stream_id, [ModelRequest([TextInput("batch-1")])])

        # Second compaction — completely new writer (as TailWindowCompact creates)
        writer2 = EventLogWriter(store)
        await writer2.persist_dropped(stream_id, [ModelRequest([TextInput("batch-2")])])

        # Final persist
        writer3 = EventLogWriter(store)
        await writer3.persist(stream_id, [ModelRequest([TextInput("final")])])

        # All three segments must be present and loadable
        assert await EventLogWriter(store).load(stream_id) == [
            ModelRequest([TextInput("batch-1")]),
            ModelRequest([TextInput("batch-2")]),
            ModelRequest([TextInput("final")]),
        ]

    async def test_persist_dropped_after_gap_uses_max_plus_one(self) -> None:
        store = MemoryKnowledgeStore()
        writer = EventLogWriter(store)
        stream_id = uuid4()

        await writer.persist_dropped(stream_id, [ModelRequest([TextInput("batch-1")])])
        await writer.persist_dropped(stream_id, [ModelRequest([TextInput("batch-2")])])

        # Simulate cleanup/retention removing the earliest segment.
        await store.delete(f"/log/{stream_id}.dropped-1.jsonl")

        await writer.persist_dropped(stream_id, [ModelRequest([TextInput("batch-3")])])
        await writer.persist(stream_id, [ModelRequest([TextInput("final")])])

        entries = await store.list("/log/")
        assert f"{stream_id}.dropped-2.jsonl" in entries
        assert f"{stream_id}.dropped-3.jsonl" in entries
        assert f"{stream_id}.dropped-1.jsonl" not in entries

        assert await writer.load(stream_id) == [
            ModelRequest([TextInput("batch-2")]),
            ModelRequest([TextInput("batch-3")]),
            ModelRequest([TextInput("final")]),
        ]

    @pytest.mark.parametrize("junk", ["abc", "1x", "²"])
    async def test_non_integer_segment_names_are_ignored(self, junk: str) -> None:
        store = MemoryKnowledgeStore()
        writer = EventLogWriter(store)
        stream_id = uuid4()

        await writer.persist_dropped(stream_id, [ModelRequest([TextInput("batch-1")])])
        await store.write(f"/log/{stream_id}.dropped-{junk}.jsonl", "not an event log")

        await writer.persist_dropped(stream_id, [ModelRequest([TextInput("batch-2")])])
        await writer.persist(stream_id, [ModelRequest([TextInput("final")])])

        assert await writer.load(stream_id) == [
            ModelRequest([TextInput("batch-1")]),
            ModelRequest([TextInput("batch-2")]),
            ModelRequest([TextInput("final")]),
        ]

    async def test_segments_load_in_numeric_order(self) -> None:
        store = MemoryKnowledgeStore()
        writer = EventLogWriter(store)
        stream_id = uuid4()

        for i in range(1, 11):
            await writer.persist_dropped(stream_id, [ModelRequest([TextInput(f"batch-{i}")])])

        assert f"{stream_id}.dropped-10.jsonl" in await store.list("/log/")
        assert await writer.load(stream_id) == [ModelRequest([TextInput(f"batch-{i}")]) for i in range(1, 11)]

    async def test_load_empty(self) -> None:
        store = MemoryKnowledgeStore()
        writer = EventLogWriter(store)
        assert await writer.load(uuid4()) == []

    async def test_unknown_event_fallback(self) -> None:
        store = MemoryKnowledgeStore()
        # Write a record with a non-existent event type
        stream_id = uuid4()
        record = json.dumps({"type": "nonexistent.module.FakeEvent", "data": {"key": "val"}})
        await store.write(f"/log/{stream_id}.jsonl", record)

        writer = EventLogWriter(store)
        assert await writer.load(stream_id) == [
            UnknownEvent(type_name="nonexistent.module.FakeEvent", data={"key": "val"})
        ]


@pytest.mark.asyncio
class TestDefaultBootstrap:
    async def test_creates_standard_layout(self) -> None:
        store = MemoryKnowledgeStore()
        # Agent writes sentinel before calling bootstrap, so simulate that
        await store.write("/.initialized", "test-agent")
        bootstrap = DefaultBootstrap()
        await bootstrap.bootstrap(store, "test-agent")

        assert await store.exists("/.initialized")
        assert await store.exists("/SKILL.md")
        assert await store.exists("/log/SKILL.md")
        assert await store.exists("/artifacts/SKILL.md")
        assert await store.exists("/memory/SKILL.md")

        root_skill = await store.read("/SKILL.md")
        assert "test-agent" in root_skill

    async def test_sentinel_prevents_rebootstrap(self) -> None:
        store = MemoryKnowledgeStore()
        # Agent writes sentinel before calling bootstrap
        await store.write("/.initialized", "agent-1")
        bootstrap = DefaultBootstrap()
        await bootstrap.bootstrap(store, "agent-1")

        # Overwrite a file
        await store.write("/SKILL.md", "custom content")

        # Sentinel exists, so a second bootstrap should be skipped by caller logic
        assert await store.exists("/.initialized")
        # (The Agent checks this before calling bootstrap)

    async def test_does_not_write_sentinel(self) -> None:
        """``DefaultBootstrap`` writes SKILL.md files but not ``/.initialized``.

        The sentinel is owned by the knowledge scope (it writes it before
        calling bootstrap, see ``ag2.knowledge._maintenance``).
        """
        store = MemoryKnowledgeStore()
        await DefaultBootstrap().bootstrap(store, "test-agent")

        assert await store.exists("/SKILL.md")
        assert not await store.exists("/.initialized")


@pytest.mark.asyncio
class TestAgentBootstrapsOnce:
    """End-to-end: a real Agent only bootstraps the store once."""

    async def test_concurrent_asks_bootstrap_once(self) -> None:
        class _CountingBootstrap:
            def __init__(self) -> None:
                self.calls = 0

            async def bootstrap(self, store, actor_name: str) -> None:
                self.calls += 1
                await store.write("/SKILL.md", f"bootstrapped by {actor_name}")

        store = MemoryKnowledgeStore()
        counter = _CountingBootstrap()
        agent = Agent(
            "booter",
            config=TestConfig(
                ModelResponse(ModelMessage("ok-1")),
                ModelResponse(ModelMessage("ok-2")),
                ModelResponse(ModelMessage("ok-3")),
            ),
            plugins=[KnowledgePlugin(store=store, bootstrap=counter)],
        )

        await asyncio.gather(
            agent.ask("hi", stream=MemoryStream()),
            agent.ask("hi", stream=MemoryStream()),
            agent.ask("hi", stream=MemoryStream()),
        )

        assert counter.calls == 1
        assert await store.exists("/.initialized")

    async def test_existing_sentinel_skips_bootstrap(self) -> None:
        class _CountingBootstrap:
            def __init__(self) -> None:
                self.calls = 0

            async def bootstrap(self, store, actor_name: str) -> None:
                self.calls += 1

        store = MemoryKnowledgeStore()
        await store.write("/.initialized", "prior-agent")

        counter = _CountingBootstrap()
        agent = Agent(
            "later",
            config=TestConfig(ModelResponse(ModelMessage("ok"))),
            plugins=[KnowledgePlugin(store=store, bootstrap=counter)],
        )
        await agent.ask("hi")

        assert counter.calls == 0


@pytest.mark.skipif(sys.platform == "win32", reason="DiskKnowledgeStore is POSIX-only")
@pytest.mark.asyncio
class TestDiskKnowledgeStore:
    async def test_read_write(self, tmp_path: Path) -> None:
        store = DiskKnowledgeStore(str(tmp_path))
        await store.write("/test.txt", "hello")
        assert await store.read("/test.txt") == "hello"
        assert (tmp_path / "test.txt").read_text() == "hello"

    async def test_persists_across_reopens(self, tmp_path: Path) -> None:
        store1 = DiskKnowledgeStore(str(tmp_path))
        await store1.write("/hello.txt", "world")

        store2 = DiskKnowledgeStore(str(tmp_path))
        assert await store2.read("/hello.txt") == "world"

    async def test_read_nonexistent(self, tmp_path: Path) -> None:
        store = DiskKnowledgeStore(str(tmp_path))
        assert await store.read("/missing.txt") is None

    async def test_write_creates_parent_directories(self, tmp_path: Path) -> None:
        store = DiskKnowledgeStore(str(tmp_path))
        await store.write("/a/b/c/deep.txt", "data")
        assert (tmp_path / "a" / "b" / "c" / "deep.txt").read_text() == "data"
        assert await store.read("/a/b/c/deep.txt") == "data"

    async def test_list_root_marks_directories(self, tmp_path: Path) -> None:
        store = DiskKnowledgeStore(str(tmp_path))
        await store.write("/a.txt", "a")
        await store.write("/sub/b.txt", "b")
        entries = await store.list("/")
        assert "a.txt" in entries
        assert "sub/" in entries

    async def test_list_nonexistent_returns_empty(self, tmp_path: Path) -> None:
        store = DiskKnowledgeStore(str(tmp_path))
        assert await store.list("/missing") == []

    async def test_delete_file(self, tmp_path: Path) -> None:
        store = DiskKnowledgeStore(str(tmp_path))
        await store.write("/test.txt", "hello")
        await store.delete("/test.txt")
        assert not (tmp_path / "test.txt").exists()
        assert await store.read("/test.txt") is None

    async def test_delete_file_prunes_empty_parent_directories(self, tmp_path: Path) -> None:
        store = DiskKnowledgeStore(str(tmp_path))
        await store.write("/dir/a.txt", "a")
        await store.delete("/dir/a.txt")
        assert await store.list("/") == []
        assert await store.exists("/dir") is False

        await store.write("/a/b/c/deep.txt", "x")
        await store.delete("/a/b/c/deep.txt")
        assert await store.list("/") == []
        assert await store.exists("/a") is False
        assert await store.exists("/a/b") is False

    async def test_delete_file_keeps_non_empty_parent_directories(self, tmp_path: Path) -> None:
        store = DiskKnowledgeStore(str(tmp_path))
        await store.write("/dir/sub/a.txt", "a")
        await store.write("/dir/b.txt", "b")
        await store.delete("/dir/sub/a.txt")
        assert await store.exists("/dir/sub") is False
        assert await store.read("/dir/b.txt") == "b"

    async def test_delete_directory_prunes_empty_parent_directories(self, tmp_path: Path) -> None:
        store = DiskKnowledgeStore(str(tmp_path))
        await store.write("/a/b/c/x.txt", "x")
        await store.delete("/a/b/c")
        assert await store.list("/") == []

    async def test_delete_never_removes_store_root(self, tmp_path: Path) -> None:
        store = DiskKnowledgeStore(str(tmp_path))
        await store.write("/a.txt", "a")
        await store.delete("/a.txt")
        assert await store.exists("/") is True
        await store.write("/b.txt", "b")
        assert await store.list("/") == ["b.txt"]

    async def test_delete_directory_recursively(self, tmp_path: Path) -> None:
        store = DiskKnowledgeStore(str(tmp_path))
        await store.write("/dir/a.txt", "a")
        await store.write("/dir/sub/b.txt", "b")
        await store.delete("/dir")
        assert not (tmp_path / "dir").exists()
        assert await store.exists("/dir") is False

    async def test_delete_missing_is_noop(self, tmp_path: Path) -> None:
        store = DiskKnowledgeStore(str(tmp_path))
        await store.delete("/never-existed.txt")  # should not raise

    async def test_exists(self, tmp_path: Path) -> None:
        store = DiskKnowledgeStore(str(tmp_path))
        await store.write("/test.txt", "hello")
        assert await store.exists("/test.txt") is True
        assert await store.exists("/missing.txt") is False

    async def test_append_returns_offset(self, tmp_path: Path) -> None:
        store = DiskKnowledgeStore(str(tmp_path))
        first = await store.append("/log.jsonl", "line1\n")
        second = await store.append("/log.jsonl", "line2\n")
        assert first == 0
        assert second == len(b"line1\n")
        assert (tmp_path / "log.jsonl").read_text() == "line1\nline2\n"

    async def test_read_range(self, tmp_path: Path) -> None:
        store = DiskKnowledgeStore(str(tmp_path))
        await store.write("/data.txt", "hello world")
        assert await store.read_range("/data.txt", 0, 5) == "hello"
        assert await store.read_range("/data.txt", 6, None) == "world"
        assert await store.read_range("/data.txt", 6) == "world"

    async def test_read_range_missing_file(self, tmp_path: Path) -> None:
        store = DiskKnowledgeStore(str(tmp_path))
        assert await store.read_range("/missing.txt", 0, 10) == ""

    async def test_path_traversal_blocked(self, tmp_path: Path) -> None:
        root = tmp_path / "root"
        root.mkdir()
        store = DiskKnowledgeStore(str(root))
        with pytest.raises(ValueError, match="Path traversal"):
            await store.write("/../escape.txt", "x")

    async def test_sibling_directory_sharing_root_prefix_blocked(self, tmp_path: Path) -> None:
        root = tmp_path / "store"
        root.mkdir()
        sibling = tmp_path / "store_evil"
        sibling.mkdir()
        (sibling / "secret.txt").write_text("secret")
        store = DiskKnowledgeStore(str(root))
        with pytest.raises(ValueError, match="Path traversal"):
            await store.read("/../store_evil/secret.txt")

    async def test_on_change_fires_on_write(self, tmp_path: Path) -> None:
        store = DiskKnowledgeStore(str(tmp_path))
        received: list[str] = []
        event = asyncio.Event()

        async def callback(path: str) -> None:
            received.append(path)
            event.set()

        sub = await store.on_change("/watched", callback)
        try:
            await store.write("/watched/file.txt", "hello")
            # watchdog delivers asynchronously via a background thread
            try:
                await asyncio.wait_for(event.wait(), timeout=5.0)
            except TimeoutError:
                pytest.skip("watchdog backend did not deliver event in time")
            assert any("file.txt" in p for p in received)
        finally:
            await sub.close()

    async def test_on_change_on_missing_file_path_keeps_it_writable(self, tmp_path: Path) -> None:
        store = DiskKnowledgeStore(str(tmp_path))

        async def callback(path: str) -> None:
            pass

        sub = await store.on_change("/memory/working.md", callback)
        try:
            assert not (tmp_path / "memory" / "working.md").exists()
            await store.write("/memory/working.md", "hello")
            assert (tmp_path / "memory" / "working.md").is_file()
            assert await store.read("/memory/working.md") == "hello"
        finally:
            await sub.close()

    async def test_on_change_on_existing_file_path_leaves_it_untouched(self, tmp_path: Path) -> None:
        store = DiskKnowledgeStore(str(tmp_path))
        await store.write("/notes.md", "original")

        async def callback(path: str) -> None:
            pass

        sub = await store.on_change("/notes.md", callback)
        try:
            assert (tmp_path / "notes.md").is_file()
            assert await store.read("/notes.md") == "original"
        finally:
            await sub.close()

    async def test_on_change_on_file_path_only_fires_for_that_file(self, tmp_path: Path) -> None:
        store = DiskKnowledgeStore(str(tmp_path))
        received: list[str] = []
        event = asyncio.Event()

        async def callback(path: str) -> None:
            received.append(path)
            if path == "/memory/working.md":
                event.set()

        sub = await store.on_change("/memory/working.md", callback)
        try:
            # sibling events arrive before the target's, so they would be in `received` by then
            await store.write("/memory/other.md", "ignored")
            await store.write("/memory/working.md.bak", "ignored")
            await store.write("/memory/working.md", "hello")
            # watchdog delivers asynchronously via a background thread
            await asyncio.wait_for(event.wait(), timeout=10.0)
            assert set(received) == {"/memory/working.md"}
        finally:
            await sub.close()


@pytest.mark.asyncio
class TestSqliteKnowledgeStore:
    async def test_read_write(self, tmp_path: Path) -> None:
        store = SqliteKnowledgeStore(str(tmp_path / "store.db"))
        try:
            await store.write("/test.txt", "hello")
            assert await store.read("/test.txt") == "hello"
        finally:
            store.close()

    async def test_read_nonexistent(self, tmp_path: Path) -> None:
        store = SqliteKnowledgeStore(str(tmp_path / "store.db"))
        try:
            assert await store.read("/missing.txt") is None
        finally:
            store.close()

    async def test_list_root_and_subdirectory(self, tmp_path: Path) -> None:
        store = SqliteKnowledgeStore(str(tmp_path / "store.db"))
        try:
            await store.write("/a.txt", "a")
            await store.write("/dir/b.txt", "b")
            await store.write("/dir/c.txt", "c")
            root = await store.list("/")
            assert "a.txt" in root
            assert "dir/" in root
            sub = await store.list("/dir/")
            assert sub == ["b.txt", "c.txt"]
        finally:
            store.close()

    async def test_delete_file_and_directory(self, tmp_path: Path) -> None:
        store = SqliteKnowledgeStore(str(tmp_path / "store.db"))
        try:
            await store.write("/file.txt", "x")
            await store.write("/dir/a.txt", "a")
            await store.write("/dir/b.txt", "b")
            await store.delete("/file.txt")
            await store.delete("/dir")
            assert await store.read("/file.txt") is None
            assert await store.read("/dir/a.txt") is None
            assert await store.read("/dir/b.txt") is None
        finally:
            store.close()

    async def test_exists_file_and_directory(self, tmp_path: Path) -> None:
        store = SqliteKnowledgeStore(str(tmp_path / "store.db"))
        try:
            await store.write("/dir/file.txt", "x")
            assert await store.exists("/dir/file.txt") is True
            assert await store.exists("/dir") is True
            assert await store.exists("/missing.txt") is False
        finally:
            store.close()

    async def test_append_returns_offset(self, tmp_path: Path) -> None:
        store = SqliteKnowledgeStore(str(tmp_path / "store.db"))
        try:
            first = await store.append("/log.jsonl", "line1\n")
            second = await store.append("/log.jsonl", "line2\n")
            assert first == 0
            assert second == len(b"line1\n")
            assert await store.read("/log.jsonl") == "line1\nline2\n"
        finally:
            store.close()

    async def test_read_range(self, tmp_path: Path) -> None:
        store = SqliteKnowledgeStore(str(tmp_path / "store.db"))
        try:
            await store.write("/data.txt", "hello world")
            assert await store.read_range("/data.txt", 0, 5) == "hello"
            assert await store.read_range("/data.txt", 6, None) == "world"
            assert await store.read_range("/data.txt", 100, 200) == ""
            assert await store.read_range("/missing.txt", 0, 10) == ""
        finally:
            store.close()

    async def test_list_versions_under_bumps_per_write(self, tmp_path: Path) -> None:
        store = SqliteKnowledgeStore(str(tmp_path / "store.db"))
        try:
            await store.write("/a.txt", "v1")
            await store.write("/dir/b.txt", "v1")
            snapshot1 = await store.list_versions_under("/")
            assert set(snapshot1.keys()) == {"/a.txt", "/dir/b.txt"}

            await store.write("/a.txt", "v2")
            snapshot2 = await store.list_versions_under("/")
            assert snapshot2["/a.txt"] > snapshot1["/a.txt"]
            assert snapshot2["/dir/b.txt"] == snapshot1["/dir/b.txt"]

            scoped = await store.list_versions_under("/dir")
            assert set(scoped.keys()) == {"/dir/b.txt"}
        finally:
            store.close()

    async def test_underscore_prefix_does_not_match_sibling(self, tmp_path: Path) -> None:
        store = SqliteKnowledgeStore(str(tmp_path / "store.db"))
        try:
            await store.write("/skills/codeXreview/notes.md", "sibling")
            assert await store.exists("/skills/code_review") is False
            assert await store.list("/skills/code_review") == []
            await store.delete("/skills/code_review")
            assert await store.read("/skills/codeXreview/notes.md") == "sibling"

            await store.write("/skills/code_review/notes.md", "real")
            assert await store.exists("/skills/code_review") is True
            assert await store.list("/skills/code_review") == ["notes.md"]
            await store.delete("/skills/code_review")
            assert await store.read("/skills/code_review/notes.md") is None
            assert await store.read("/skills/codeXreview/notes.md") == "sibling"
        finally:
            store.close()

    async def test_percent_prefix_does_not_match_sibling(self, tmp_path: Path) -> None:
        store = SqliteKnowledgeStore(str(tmp_path / "store.db"))
        try:
            await store.write("/pct/100X_done/sibling.md", "sibling")
            assert await store.list("/pct/100%_done") == []
            assert await store.exists("/pct/100%_done") is False
            await store.delete("/pct/100%_done")
            assert await store.read("/pct/100X_done/sibling.md") == "sibling"

            await store.write("/pct/100%_done/real.md", "real")
            assert await store.list("/pct/100%_done") == ["real.md"]
        finally:
            store.close()

    async def test_prefix_match_is_case_sensitive(self, tmp_path: Path) -> None:
        store = SqliteKnowledgeStore(str(tmp_path / "store.db"))
        try:
            await store.write("/docs/readme.md", "hello")
            assert await store.list("/Docs") == []
            assert await store.exists("/Docs") is False
            await store.delete("/Docs")
            assert await store.read("/docs/readme.md") == "hello"
            assert await store.list("/docs") == ["readme.md"]
        finally:
            store.close()

    @pytest.mark.parametrize(
        ("prefix", "real", "sibling"),
        [
            ("/skills/code_review", "/skills/code_review/notes.md", "/skills/codeXreview/notes.md"),
            ("/pct/100%_done", "/pct/100%_done/notes.md", "/pct/100X_done/notes.md"),
            ("/docs", "/docs/notes.md", "/Docs/notes.md"),
            ("/café_menu", "/café_menu/notes.md", "/caféXmenu/notes.md"),
        ],
    )
    async def test_list_versions_under_matches_prefix_literally(
        self, tmp_path: Path, prefix: str, real: str, sibling: str
    ) -> None:
        store = SqliteKnowledgeStore(str(tmp_path / "store.db"))
        try:
            await store.write(real, "real")
            await store.write(sibling, "sibling")
            versions = await store.list_versions_under(prefix)
            assert set(versions.keys()) == {real}
        finally:
            store.close()

    async def test_root_prefix_lists_every_entry(self, tmp_path: Path) -> None:
        store = SqliteKnowledgeStore(str(tmp_path / "store.db"))
        try:
            await store.write("/a_b/one.md", "1")
            await store.write("/aXb/two.md", "2")
            assert await store.list("/") == ["aXb/", "a_b/"]
            assert set((await store.list_versions_under("/")).keys()) == {"/a_b/one.md", "/aXb/two.md"}
        finally:
            store.close()

    async def test_on_change_polling(self, tmp_path: Path) -> None:
        store = SqliteKnowledgeStore(str(tmp_path / "store.db"), poll_interval_s=0.05)
        received: list[str] = []
        event = asyncio.Event()

        async def callback(path: str) -> None:
            received.append(path)
            event.set()

        sub = await store.on_change("/watched", callback)
        try:
            await store.write("/watched/file.txt", "hello")
            await asyncio.wait_for(event.wait(), timeout=2.0)
            assert "/watched/file.txt" in received
        finally:
            await sub.close()
            store.close()

    async def test_close_is_idempotent(self, tmp_path: Path) -> None:
        store = SqliteKnowledgeStore(str(tmp_path / "store.db"))
        store.close()
        store.close()  # should not raise

    async def test_persistence_across_instances(self, tmp_path: Path) -> None:
        db_path = str(tmp_path / "store.db")
        store1 = SqliteKnowledgeStore(db_path)
        await store1.write("/persisted.txt", "survives")
        store1.close()

        store2 = SqliteKnowledgeStore(db_path)
        try:
            assert await store2.read("/persisted.txt") == "survives"
        finally:
            store2.close()

    async def test_concurrent_reads_and_writes_do_not_corrupt_connection(self, tmp_path: Path) -> None:
        """Interleaved reads and writes share one connection and must serialize.

        The single ``sqlite3.connect(check_same_thread=False)`` connection is
        dispatched across the default executor's thread pool. Without the
        ``asyncio.Lock`` on the read paths a concurrent read and write land on
        the connection at the same time and sqlite3 misbehaves — either raising
        ``sqlite3.InterfaceError: bad parameter or other API misuse`` or leaving
        a cursor wedged so sibling executor threads block forever. Hammer the
        store with ~200 interleaved operations on one instance to catch it.

        The whole storm is wrapped in ``asyncio.wait_for`` so a corrupted
        connection surfaces as a deterministic failure (``TimeoutError`` or the
        raised ``InterfaceError``) instead of hanging the suite.
        """
        store = SqliteKnowledgeStore(str(tmp_path / "store.db"))
        try:
            ops = []
            for i in range(50):
                path = f"/t/{i}"
                ops.append(store.write(path, f"value-{i}"))
                ops.append(store.read(path))
                ops.append(store.exists(path))
                ops.append(store.list("/t/"))
            # gather does not swallow exceptions here; an InterfaceError from
            # any interleaving propagates and fails the test. wait_for guards
            # against the wedged-connection hang the bug can also produce.
            await asyncio.wait_for(asyncio.gather(*ops), timeout=15.0)

            # Store is still consistent after the storm.
            assert await store.read("/t/0") == "value-0"
            assert await store.read("/t/49") == "value-49"
            assert await store.exists("/t/25") is True
            listing = await store.list("/t/")
            assert len(listing) == 50
        finally:
            store.close()


class _FakeLock:
    """Minimal Lock protocol implementation for LockedKnowledgeStore tests."""

    def __init__(self) -> None:
        self.acquired: list[tuple[str, float]] = []
        self.released: list[str] = []
        self.acquire_result = True

    async def acquire(self, name: str, ttl: float) -> bool:
        self.acquired.append((name, ttl))
        return self.acquire_result

    async def release(self, name: str) -> None:
        self.released.append(name)


@pytest.mark.asyncio
class TestLockedKnowledgeStore:
    async def test_read_bypasses_lock(self) -> None:
        inner = MemoryKnowledgeStore()
        await inner.write("/test.txt", "data")
        lock = _FakeLock()
        locked = LockedKnowledgeStore(inner, lock)

        assert await locked.read("/test.txt") == "data"
        assert lock.acquired == []

    async def test_write_acquires_and_releases(self) -> None:
        inner = MemoryKnowledgeStore()
        lock = _FakeLock()
        locked = LockedKnowledgeStore(inner, lock)

        await locked.write("/test.txt", "hello")

        assert lock.acquired == [("store:write:/test.txt", 30.0)]
        assert lock.released == ["store:write:/test.txt"]
        assert await inner.read("/test.txt") == "hello"

    async def test_write_raises_when_lock_unavailable(self) -> None:
        inner = MemoryKnowledgeStore()
        lock = _FakeLock()
        lock.acquire_result = False
        locked = LockedKnowledgeStore(inner, lock)

        with pytest.raises(RuntimeError, match="Failed to acquire write lock"):
            await locked.write("/test.txt", "hello")
        assert lock.released == []
        assert await inner.read("/test.txt") is None

    async def test_write_releases_on_inner_exception(self) -> None:
        class ExplodingStore:
            async def write(self, path: str, content: str) -> None:
                raise ValueError("boom")

        lock = _FakeLock()
        locked = LockedKnowledgeStore(ExplodingStore(), lock)  # type: ignore[arg-type]

        with pytest.raises(ValueError, match="boom"):
            await locked.write("/test.txt", "hello")
        assert lock.released == ["store:write:/test.txt"]

    async def test_delete_acquires_and_releases(self) -> None:
        inner = MemoryKnowledgeStore()
        await inner.write("/test.txt", "data")
        lock = _FakeLock()
        locked = LockedKnowledgeStore(inner, lock)

        await locked.delete("/test.txt")
        assert lock.acquired == [("store:write:/test.txt", 30.0)]
        assert lock.released == ["store:write:/test.txt"]
        assert await inner.read("/test.txt") is None

    async def test_delete_raises_when_lock_unavailable(self) -> None:
        inner = MemoryKnowledgeStore()
        lock = _FakeLock()
        lock.acquire_result = False
        locked = LockedKnowledgeStore(inner, lock)

        with pytest.raises(RuntimeError, match="Failed to acquire delete lock"):
            await locked.delete("/test.txt")

    async def test_append_acquires_and_returns_offset(self) -> None:
        inner = MemoryKnowledgeStore()
        lock = _FakeLock()
        locked = LockedKnowledgeStore(inner, lock)

        first = await locked.append("/log.jsonl", "line1\n")
        second = await locked.append("/log.jsonl", "line2\n")

        assert first == 0
        assert second == len(b"line1\n")
        assert lock.acquired == [
            ("store:write:/log.jsonl", 30.0),
            ("store:write:/log.jsonl", 30.0),
        ]
        assert lock.released == [
            "store:write:/log.jsonl",
            "store:write:/log.jsonl",
        ]

    async def test_append_raises_when_lock_unavailable(self) -> None:
        inner = MemoryKnowledgeStore()
        lock = _FakeLock()
        lock.acquire_result = False
        locked = LockedKnowledgeStore(inner, lock)

        with pytest.raises(RuntimeError, match="Failed to acquire append lock"):
            await locked.append("/log.jsonl", "line1\n")

    async def test_read_list_exists_read_range_bypass_lock(self) -> None:
        inner = MemoryKnowledgeStore()
        await inner.write("/dir/a.txt", "hello")
        lock = _FakeLock()
        locked = LockedKnowledgeStore(inner, lock)

        assert await locked.list("/dir") == ["a.txt"]
        assert await locked.exists("/dir/a.txt") is True
        assert await locked.read_range("/dir/a.txt", 0, 5) == "hello"
        assert lock.acquired == []

    async def test_on_change_bypasses_lock(self) -> None:
        inner = MemoryKnowledgeStore()
        lock = _FakeLock()
        locked = LockedKnowledgeStore(inner, lock)
        received: list[str] = []

        async def callback(path: str) -> None:
            received.append(path)

        sub = await locked.on_change("/watched", callback)
        try:
            await inner.write("/watched/file.txt", "x")
            assert received == ["/watched/file.txt"]
            assert lock.acquired == []
        finally:
            await sub.close()

    @pytest.mark.parametrize("alias", ["log/events.jsonl", "//log//events.jsonl/", "/log/events.jsonl/"])
    async def test_aliased_path_takes_the_canonical_lock(self, alias: str) -> None:
        lock = _FakeLock()
        locked = LockedKnowledgeStore(MemoryKnowledgeStore(), lock)

        await locked.write(alias, "hello")
        await locked.append(alias, " world")
        await locked.delete(alias)

        canonical = "store:write:/log/events.jsonl"
        assert lock.acquired == [(canonical, 30.0)] * 3
        assert lock.released == [canonical] * 3

    @pytest.mark.parametrize("alias", ["log/events.jsonl", "//log//events.jsonl/", "/log/events.jsonl/"])
    async def test_aliased_path_releases_the_canonical_lock_on_inner_exception(self, alias: str) -> None:
        class ExplodingStore:
            async def write(self, path: str, content: str) -> None:
                raise ValueError("boom")

        lock = _FakeLock()
        locked = LockedKnowledgeStore(ExplodingStore(), lock)  # type: ignore[arg-type]

        with pytest.raises(ValueError, match="boom"):
            await locked.write(alias, "hello")
        assert lock.released == ["store:write:/log/events.jsonl"]