"""Engine tests with fake processors (no LLM): flow, barrier, retry/dead-letter, producer race."""

import asyncio

import pytest

import kanban
from kanban import Flow, Stage, chain_stages, run_flow

TOPIC = "t"
BOARD = "inventory"


def _advance_proc(db, to_stage):
    async def proc(cards):
        await db.kanban_advance_many(TOPIC, BOARD, [(c["card_id"], to_stage) for c in cards])
    return proc


async def _seed(db, n, stage="s1"):
    for i in range(n):
        await db.kanban_upsert_card(TOPIC, BOARD, f"card-{i}", "title", stage, {"title": f"T{i}"})


async def test_cards_flow_through_stages(testdb):
    db = testdb
    await _seed(db, 7)
    flow = Flow(TOPIC)
    stages = chain_stages([
        Stage(BOARD, "s1", _advance_proc(db, "s2")),
        Stage(BOARD, "s2", _advance_proc(db, "done")),
    ])
    await asyncio.wait_for(run_flow(flow, stages), timeout=10)
    assert await db.kanban_count(TOPIC, "done", board=BOARD) == 7
    assert await db.kanban_count(TOPIC, ["s1", "s2"], board=BOARD) == 0


async def test_barrier_waits_for_upstream(testdb):
    db = testdb
    await _seed(db, 6)
    upstream_left: list[int] = []

    async def slow_s1(cards):
        await asyncio.sleep(0.05)  # keep upstream busy so an eager barrier would see queued cards
        await db.kanban_advance_many(TOPIC, BOARD, [(c["card_id"], "gate") for c in cards])

    async def barrier_proc(cards):
        upstream_left.append(await db.kanban_count(TOPIC, ["s1"], board=BOARD))
        await db.kanban_advance_many(TOPIC, BOARD, [(c["card_id"], "done") for c in cards])

    flow = Flow(TOPIC)
    stages = chain_stages([
        Stage(BOARD, "s1", slow_s1),
        Stage(BOARD, "gate", barrier_proc, barrier=True),
    ])
    await asyncio.wait_for(run_flow(flow, stages), timeout=10)
    assert await db.kanban_count(TOPIC, "done", board=BOARD) == 6
    assert upstream_left and all(n == 0 for n in upstream_left)  # barrier never ran with s1 queued


async def test_retry_backoff_then_dead(testdb, monkeypatch):
    db = testdb
    monkeypatch.setattr(kanban, "RETRY_BACKOFF", 0.02)
    await _seed(db, 1)
    attempts = []

    async def failing(cards):
        attempts.append(cards[0]["retries"])
        raise RuntimeError("kaputt")

    flow = Flow(TOPIC)
    stages = chain_stages([Stage(BOARD, "s1", failing)])
    await asyncio.wait_for(run_flow(flow, stages), timeout=10)
    card = await db.kanban_get_card(TOPIC, BOARD, "card-0")
    assert card["stage"] == "dead"
    assert card["retries"] == kanban.MAX_CARD_RETRIES
    assert "kaputt" in card["last_error"]
    assert attempts == [0, 1, 2]  # backoff between attempts, then dead-letter


async def test_requeue_dead(testdb):
    db = testdb
    await db.kanban_upsert_card(TOPIC, BOARD, "card-0", "title", "s1")
    for _ in range(kanban.MAX_CARD_RETRIES):
        await db.kanban_fail_card(TOPIC, BOARD, "card-0", "x", kanban.MAX_CARD_RETRIES, 0.0)
    assert (await db.kanban_get_card(TOPIC, BOARD, "card-0"))["stage"] == "dead"
    assert await db.kanban_requeue_dead(TOPIC, BOARD, "s1") == 1
    card = await db.kanban_get_card(TOPIC, BOARD, "card-0")
    assert card["stage"] == "s1" and card["retries"] == 0


async def test_producer_attach_in_idle_lull(testdb):
    """Fix-6 regression: a producer attached while workers sit in the exit grace poll
    must keep the flow alive and its cards must still be processed."""
    db = testdb
    flow = Flow(TOPIC)
    stages = chain_stages([Stage(BOARD, "s1", _advance_proc(db, "done"))])

    async def producer_a():
        await db.kanban_upsert_card(TOPIC, BOARD, "card-a", "title", "s1")
        flow.wake.set()
        flow.done_producer()

    async def attacher():
        while await db.kanban_count(TOPIC, "done", board=BOARD) == 0:  # wait for card-a done
            await asyncio.sleep(0.01)
        flow.add_producer()  # synchronous BEFORE the work — the grace poll must see it

        async def producer_b():
            await db.kanban_upsert_card(TOPIC, BOARD, "card-b", "title", "s1")
            flow.wake.set()
            flow.done_producer()
        await producer_b()

    flow.add_producer()  # producer_a, counted before run_flow (sync add)
    asyncio.get_event_loop().create_task(attacher())
    await asyncio.wait_for(run_flow(flow, stages, producers=[producer_a()]), timeout=10)
    assert await db.kanban_count(TOPIC, "done", board=BOARD) == 2


async def test_backoff_delays_pull(testdb, monkeypatch):
    db = testdb
    await db.kanban_upsert_card(TOPIC, BOARD, "card-0", "title", "s1")
    await db.kanban_fail_card(TOPIC, BOARD, "card-0", "x", 5, 0.2)
    assert await db.kanban_pull(TOPIC, BOARD, "s1", 10) == []   # in backoff → not pullable
    assert await db.kanban_count(TOPIC, "s1", board=BOARD) == 1  # but still counts as queued
    await asyncio.sleep(0.25)
    assert len(await db.kanban_pull(TOPIC, BOARD, "s1", 10)) == 1