"""New threads must not scan legacy SQLite sources on empty primary reads."""

import asyncio
from types import SimpleNamespace

from deerflow.runtime.checkpointer.checkpoint_migration import DualReadCheckpointer


class _Primary:
    def __init__(self):
        self.tuple_calls = 0
        self.list_calls = 0

    async def aget_tuple(self, config):
        self.tuple_calls += 1
        return None

    async def aget(self, config):
        return None

    async def alist(self, config, *, filter=None, before=None, limit=None):
        self.list_calls += 1
        if False:
            yield None

    async def aput(self, *args, **kwargs):
        return None


class _Legacy:
    def __init__(self):
        self.tuple_calls = 0
        self.list_calls = 0
        self.closed = False

    async def aget_tuple(self, config):
        self.tuple_calls += 1
        return {"legacy": True}

    async def alist(self, config, *, filter=None, before=None, limit=None):
        self.list_calls += 1
        if False:
            yield None


def test_remembered_thread_skips_empty_sqlite_fallback():
    async def run():
        primary = _Primary()
        legacy = _Legacy()
        saver = DualReadCheckpointer(primary, [SimpleNamespace(saver=legacy, closed=False)])
        config = {"configurable": {"thread_id": "new-hello"}}

        assert await saver.aget_tuple(config) == {"legacy": True}
        assert legacy.tuple_calls == 1

        saver.remember_primary_thread("new-hello")
        assert await saver.aget_tuple(config) is None
        assert legacy.tuple_calls == 1

        chunks = [item async for item in saver.alist(config, limit=1)]
        assert chunks == []
        assert legacy.list_calls == 0

        await saver.aput(config, {}, {}, {})
        assert saver._skip_legacy_fallback(config) is True

    asyncio.run(run())