import json

import aiosqlite
from config import DB_PATH

CREATE_GUIDES = """
CREATE TABLE IF NOT EXISTS guides (
    id TEXT PRIMARY KEY,
    topic TEXT NOT NULL,
    format TEXT NOT NULL,
    instructions TEXT NOT NULL DEFAULT '',
    status TEXT NOT NULL DEFAULT 'queued',
    progress TEXT,
    step INTEGER,
    error_msg TEXT,
    created_at TEXT NOT NULL,
    updated_at TEXT NOT NULL
)
"""

CREATE_TOPICS = """
CREATE TABLE IF NOT EXISTS topics (
    name TEXT PRIMARY KEY,
    created_at TEXT NOT NULL
)
"""

CREATE_BLOCK_PROGRESS = """
CREATE TABLE IF NOT EXISTS block_progress (
    topic TEXT NOT NULL,
    block TEXT NOT NULL,
    good_answers INTEGER NOT NULL DEFAULT 0,
    streak INTEGER NOT NULL DEFAULT 0,
    completed TEXT,
    understood TEXT,
    mastered TEXT,
    updated_at TEXT NOT NULL,
    PRIMARY KEY (topic, block)
)
"""

# --- Blocks pipeline content (replaces file sidecars) ---

# Inventory: one block per (topic, title_norm). mentions = number of agents/rounds
# that named it (≥2 = consensus). status: candidate/consensus/rest/discarded.
CREATE_BLOCKS = """
CREATE TABLE IF NOT EXISTS blocks (
    topic TEXT NOT NULL,
    title_norm TEXT NOT NULL,
    title TEXT NOT NULL,
    description TEXT NOT NULL DEFAULT '',
    mentions INTEGER NOT NULL DEFAULT 1,
    status TEXT NOT NULL DEFAULT 'candidate',
    sources TEXT NOT NULL DEFAULT '[]',
    reader TEXT NOT NULL DEFAULT '[]',
    updated_at TEXT NOT NULL,
    PRIMARY KEY (topic, title_norm)
)
"""

# Subblocks per block. level (einfach/mittel/schwer) + relevance (relevant/rand)
# are set later. mentions analogous to the inventory.
CREATE_SUBBLOCKS = """
CREATE TABLE IF NOT EXISTS subblocks (
    topic TEXT NOT NULL,
    block_norm TEXT NOT NULL,
    sub_norm TEXT NOT NULL,
    block TEXT NOT NULL,
    sub_title TEXT NOT NULL,
    mentions INTEGER NOT NULL DEFAULT 1,
    level TEXT,
    relevance TEXT,
    facts TEXT NOT NULL DEFAULT '',
    status TEXT NOT NULL DEFAULT 'candidate',
    updated_at TEXT NOT NULL,
    PRIMARY KEY (topic, block_norm, sub_norm)
)
"""

# One question pattern per (block, subblock). Difficulty comes only at exam time
# from the learner tier, not from the pattern — hence no more type cross-product.
CREATE_QUESTION_PATTERN = """
CREATE TABLE IF NOT EXISTS question_pattern (
    topic TEXT NOT NULL,
    block_norm TEXT NOT NULL,
    sub_norm TEXT NOT NULL,
    block TEXT NOT NULL,
    sub_title TEXT NOT NULL,
    question TEXT NOT NULL,
    updated_at TEXT NOT NULL,
    PRIMARY KEY (topic, block_norm, sub_norm)
)
"""

# Crawl pages per topic: content (1=content/0=noise, from the triage) + read_done (from the research).
CREATE_RESEARCH_COVERAGE = """
CREATE TABLE IF NOT EXISTS research_coverage (
    topic TEXT NOT NULL,
    source TEXT NOT NULL,
    read_done INTEGER NOT NULL DEFAULT 0,
    content INTEGER,
    updated_at TEXT NOT NULL,
    PRIMARY KEY (topic, source)
)
"""

# Step status of the pipeline (replaces file-existence resume + reset globs).
CREATE_PIPELINE_STATE = """
CREATE TABLE IF NOT EXISTS pipeline_state (
    topic TEXT NOT NULL,
    step TEXT NOT NULL,
    status TEXT NOT NULL DEFAULT 'open',
    updated_at TEXT NOT NULL,
    PRIMARY KEY (topic, step)
)
"""

# Finished guide content per (topic, format) as a JSON blob (replaces the guide JSON file).
# Shared across all guide runs of the same topic+format (as the content file was before).
CREATE_GUIDE_CONTENT = """
CREATE TABLE IF NOT EXISTS guide_content (
    topic TEXT NOT NULL,
    format TEXT NOT NULL,
    json TEXT NOT NULL,
    updated_at TEXT NOT NULL,
    PRIMARY KEY (topic, format)
)
"""

# Source choice per topic (replaces source.json).
CREATE_SOURCE = """
CREATE TABLE IF NOT EXISTS source (
    topic TEXT PRIMARY KEY,
    type TEXT NOT NULL,
    location TEXT NOT NULL DEFAULT '',
    spec TEXT NOT NULL DEFAULT '',
    updated_at TEXT NOT NULL
)
"""

# Outline (chapter → block numbers) per topic as JSON. Produced in the blocks phase,
# the guide only reads it + filters per format. Format-agnostic (all blocks).
CREATE_GUIDE_OUTLINE = """
CREATE TABLE IF NOT EXISTS guide_outline (
    topic TEXT PRIMARY KEY,
    json TEXT NOT NULL,
    updated_at TEXT NOT NULL
)
"""

# Generic learning artifact layer: one artifact per (block, subblock, type) (JSON in `data`).
# type: flashcard | example. sub_norm='' = block level (for future block-wide artifacts).
# Produced from the facts in the blocks phase, presented by the frontend (artifact/display split).
CREATE_SUB_ARTEFAKTE = """
CREATE TABLE IF NOT EXISTS sub_artefakte (
    topic TEXT NOT NULL,
    block_norm TEXT NOT NULL,
    sub_norm TEXT NOT NULL DEFAULT '',
    type TEXT NOT NULL,
    block TEXT NOT NULL DEFAULT '',
    sub_title TEXT NOT NULL DEFAULT '',
    data TEXT NOT NULL DEFAULT '{}',
    updated_at TEXT NOT NULL,
    PRIMARY KEY (topic, block_norm, sub_norm, type)
)
"""

# Leitner learning state per flashcard — IDENTITY-keyed (no content): survives the
# sub_artefakte wipe on regeneration; orphaned rows simply never match in the deck join.
CREATE_PRACTICE_PROGRESS = """
CREATE TABLE IF NOT EXISTS practice_progress (
    topic TEXT NOT NULL,
    block_norm TEXT NOT NULL,
    sub_norm TEXT NOT NULL,
    box INTEGER NOT NULL DEFAULT 1,
    due_at TEXT NOT NULL,
    updated_at TEXT NOT NULL,
    PRIMARY KEY (topic, block_norm, sub_norm)
)
"""

# Kanban dataflow (boards 'inventory' + 'artefacts'): ONE generic card table for all card kinds
# (title/cluster/block). `stage` is the queue key — a worker pulls WHERE stage = <its input stage>.
# `payload` is a JSON blob (title, description, sources, readers, mentions, parent_norm, journal …);
# retries/not_before/last_error implement backoff + dead-letter (stage 'dead' after MAX_CARD_RETRIES).
CREATE_KANBAN_CARDS = """
CREATE TABLE IF NOT EXISTS kanban_cards (
    topic TEXT NOT NULL,
    board TEXT NOT NULL,
    card_id TEXT NOT NULL,
    kind TEXT NOT NULL,
    stage TEXT NOT NULL,
    payload TEXT NOT NULL DEFAULT '{}',
    retries INTEGER NOT NULL DEFAULT 0,
    not_before TEXT NOT NULL DEFAULT '',
    last_error TEXT,
    updated_at TEXT NOT NULL,
    PRIMARY KEY (topic, board, card_id)
)
"""

# Pipeline history (agents, stage moves, failures) — the raw data for quality/perf
# analysis. Written fire-and-forget from the hooks in kanban_advance_many / kanban_fail_card /
# set_guide_card and agents.run_agent (injected via agents.on_event); never load-bearing.
CREATE_EVENTS = """
CREATE TABLE IF NOT EXISTS events (
    id INTEGER PRIMARY KEY,
    topic TEXT NOT NULL,
    ts TEXT NOT NULL,
    kind TEXT NOT NULL,
    key TEXT NOT NULL DEFAULT '',
    label TEXT NOT NULL DEFAULT '',
    status TEXT NOT NULL DEFAULT '',
    dur_ms INTEGER,
    wait_ms INTEGER,
    meta TEXT NOT NULL DEFAULT '{}'
)
"""

CREATE_EVENTS_INDEX = """
CREATE INDEX IF NOT EXISTS idx_events ON events(topic, ts)
"""

CREATE_KANBAN_PULL_INDEX = """
CREATE INDEX IF NOT EXISTS idx_kanban_pull ON kanban_cards(topic, board, stage, not_before, updated_at)
"""

# title_norm → cluster membership (one title belongs to exactly one cluster).
CREATE_KANBAN_MEMBERS = """
CREATE TABLE IF NOT EXISTS kanban_members (
    topic TEXT NOT NULL,
    member_id TEXT NOT NULL,
    group_id TEXT NOT NULL,
    PRIMARY KEY (topic, member_id)
)
"""

# Guide board: one card per block, linear stages (lernziele … lesbarkeit → done).
# `md` carries the writer fragment (with kapitel/section/sub markers) between the gates.
CREATE_GUIDE_CARDS = """
CREATE TABLE IF NOT EXISTS guide_cards (
    topic TEXT NOT NULL,
    format TEXT NOT NULL DEFAULT 'Guide',
    block_norm TEXT NOT NULL,
    block TEXT NOT NULL,
    chapter TEXT NOT NULL DEFAULT '',
    ord INTEGER NOT NULL DEFAULT 0,
    stage TEXT NOT NULL DEFAULT 'lernziele',
    status TEXT NOT NULL DEFAULT 'open',
    writer_rounds INTEGER NOT NULL DEFAULT 0,
    gate_info TEXT NOT NULL DEFAULT '',
    md TEXT NOT NULL DEFAULT '',
    updated_at TEXT NOT NULL,
    PRIMARY KEY (topic, format, block_norm)
)
"""

CREATE_GUIDE_LERNZIELE = """
CREATE TABLE IF NOT EXISTS guide_lernziele (
    topic TEXT NOT NULL,
    block_norm TEXT NOT NULL,
    ziel_id TEXT NOT NULL,
    text TEXT NOT NULL,
    sub_norm TEXT NOT NULL DEFAULT '',
    covered INTEGER NOT NULL DEFAULT 0,
    updated_at TEXT NOT NULL,
    PRIMARY KEY (topic, block_norm, ziel_id)
)
"""

_db: aiosqlite.Connection | None = None


async def get_db() -> aiosqlite.Connection:
    global _db
    if _db is None:
        _db = await aiosqlite.connect(DB_PATH)
        _db.row_factory = None
    return _db


async def init_db():
    db = await get_db()
    # WAL survives crashes much better; busy_timeout absorbs short locks.
    # synchronous=NORMAL: safe under WAL, much less fsync — the kanban flow commits often.
    await db.execute("PRAGMA journal_mode=WAL")
    await db.execute("PRAGMA synchronous=NORMAL")
    await db.execute("PRAGMA busy_timeout=5000")
    await db.execute(CREATE_GUIDES)
    await db.execute(CREATE_TOPICS)
    await db.execute(CREATE_BLOCK_PROGRESS)
    await db.execute(CREATE_BLOCKS)
    await db.execute(CREATE_SUBBLOCKS)
    await db.execute(CREATE_QUESTION_PATTERN)
    await db.execute(CREATE_RESEARCH_COVERAGE)
    await db.execute(CREATE_PIPELINE_STATE)
    await db.execute(CREATE_GUIDE_CONTENT)
    await db.execute(CREATE_SOURCE)
    await db.execute(CREATE_GUIDE_OUTLINE)
    await db.execute(CREATE_SUB_ARTEFAKTE)
    await db.execute(CREATE_PRACTICE_PROGRESS)
    await db.execute(CREATE_KANBAN_CARDS)
    await db.execute(CREATE_EVENTS)
    await db.execute(CREATE_EVENTS_INDEX)
    await db.execute(CREATE_KANBAN_PULL_INDEX)
    await db.execute(CREATE_KANBAN_MEMBERS)
    await db.execute(CREATE_GUIDE_CARDS)
    await db.execute(CREATE_GUIDE_LERNZIELE)
    try:  # migration for existing DBs without the step column
        await db.execute("ALTER TABLE guides ADD COLUMN step INTEGER")
    except aiosqlite.OperationalError:
        pass
    try:  # migration: research_coverage.content (content/noise from the triage)
        await db.execute("ALTER TABLE research_coverage ADD COLUMN content INTEGER")
    except aiosqlite.OperationalError:
        pass
    try:  # migration for existing DBs without the understood column (mastery level)
        await db.execute("ALTER TABLE block_progress ADD COLUMN understood TEXT")
    except aiosqlite.OperationalError:
        pass
    try:  # migration for existing DBs without the mastered column (master path 25)
        await db.execute("ALTER TABLE block_progress ADD COLUMN mastered TEXT")
    except aiosqlite.OperationalError:
        pass
    try:  # migration for existing DBs without the streak column (persistent streak-bonus run)
        await db.execute("ALTER TABLE block_progress ADD COLUMN streak INTEGER NOT NULL DEFAULT 0")
    except aiosqlite.OperationalError:
        pass
    # Open-question anchor: base/streak BEFORE the currently open question — makes the rating
    # idempotent server-side (re-rating) and drift-free across questions.
    for _col, _type in (("offene_question", "TEXT"), ("offene_basis", "INTEGER"), ("offene_streak", "INTEGER")):
        try:
            await db.execute(f"ALTER TABLE block_progress ADD COLUMN {_col} {_type}")
        except aiosqlite.OperationalError:
            pass
    # Migration: question_pattern without a type column (1 pattern per sub instead of a sub×type cross-product).
    # PK change → rebuild the table once. Existing patterns are lost (intentional, no mapping).
    cursor = await db.execute("PRAGMA table_info(question_pattern)")
    if any(_r[1] == "type" for _r in await cursor.fetchall()):
        await db.execute("DROP TABLE question_pattern")
        await db.execute(CREATE_QUESTION_PATTERN)
    try:  # migration: subblocks.facts (source facts per sub, JSON blob) — extract-once grounding.
        await db.execute("ALTER TABLE subblocks ADD COLUMN facts TEXT NOT NULL DEFAULT ''")  # DEFAULT '' needed for NOT NULL on ADD COLUMN
    except aiosqlite.OperationalError:
        pass
    try:  # migration: blocks.reader (reader set per candidate, JSON) — exact consensus count (≥2 readers)
        await db.execute("ALTER TABLE blocks ADD COLUMN reader TEXT NOT NULL DEFAULT '[]'")
    except aiosqlite.OperationalError:
        pass
    # Migration: removed features leave orphaned tables behind — drop them.
    # elements (feature removed), vertiefungen/block_texte (never read), guide_progress
    # (chapter progress had no frontend and its only reader ignored the value).
    await db.execute("DROP TABLE IF EXISTS elements")
    await db.execute("DROP TABLE IF EXISTS vertiefungen")
    await db.execute("DROP TABLE IF EXISTS block_texte")
    await db.execute("DROP TABLE IF EXISTS guide_progress")
    await db.execute(
        "UPDATE guides SET status = 'error', progress = NULL, error_msg = 'Server restart' "
        "WHERE status IN ('queued', 'generating')"
    )
    await db.commit()


async def close_db():
    global _db
    if _db is not None:
        await _db.close()
        _db = None


def _row_to_dict(row, cursor):
    columns = [d[0] for d in cursor.description]
    return dict(zip(columns, row))


async def create_guide(guide: dict) -> dict:
    db = await get_db()
    await db.execute(
        """INSERT INTO guides (id, topic, format, instructions, status, progress, created_at, updated_at)
           VALUES (:id, :topic, :format, :instructions, :status, :progress, :created_at, :updated_at)""",
        guide,
    )
    await db.commit()
    return guide


async def get_guide(guide_id: str) -> dict | None:
    db = await get_db()
    cursor = await db.execute("SELECT * FROM guides WHERE id = ?", (guide_id,))
    row = await cursor.fetchone()
    if row is None:
        return None
    return _row_to_dict(row, cursor)


async def list_guides() -> list[dict]:
    db = await get_db()
    cursor = await db.execute("SELECT * FROM guides ORDER BY created_at DESC")
    rows = await cursor.fetchall()
    return [_row_to_dict(row, cursor) for row in rows]


async def _update(table: str, fields: dict, where: dict) -> None:
    """UPDATE <table> SET <fields> WHERE <where> (+ commit). WHERE params are aliased (`w_<k>`)
    so a field and a WHERE key of the same name don't collide — needed e.g. for a `title_norm`
    rename (SET new norm WHERE old norm)."""
    sets = ", ".join(f"{k} = :{k}" for k in fields)
    cond = " AND ".join(f"{k} = :w_{k}" for k in where)
    db = await get_db()
    await db.execute(f"UPDATE {table} SET {sets} WHERE {cond}", {**fields, **{f"w_{k}": v for k, v in where.items()}})
    await db.commit()


async def update_guide(guide_id: str, **fields) -> None:
    await _update("guides", fields, {"id": guide_id})


async def delete_guide(guide_id: str) -> bool:
    db = await get_db()
    cursor = await db.execute("DELETE FROM guides WHERE id = ?", (guide_id,))
    await db.commit()
    return cursor.rowcount > 0


# --- Topics ---

async def create_topic(name: str) -> None:
    from datetime import datetime, timezone
    db = await get_db()
    await db.execute(
        "INSERT OR IGNORE INTO topics (name, created_at) VALUES (?, ?)",
        (name, datetime.now(timezone.utc).isoformat()),
    )
    await db.commit()


async def list_topics() -> list[str]:
    db = await get_db()
    cursor = await db.execute("SELECT name FROM topics ORDER BY created_at DESC")
    rows = await cursor.fetchall()
    return [row[0] for row in rows]


async def delete_topic(name: str) -> None:
    db = await get_db()
    await db.execute("DELETE FROM topics WHERE name = ?", (name,))
    await db.commit()


# --- Block learning: deep-dives + exam progress ---

def _now() -> str:
    from datetime import datetime, timezone
    return datetime.now(timezone.utc).isoformat()


async def list_block_progress(topic: str) -> list[dict]:
    db = await get_db()
    cursor = await db.execute(
        "SELECT block, good_answers, streak, completed, understood, mastered FROM block_progress WHERE topic = ?", (topic,)
    )
    rows = await cursor.fetchall()
    return [{"block": b, "good_answers": n, "streak": s, "completed": a, "understood": v, "mastered": m} for b, n, s, a, v, m in rows]


async def get_block_progress(topic: str, block: str) -> dict:
    """One block row incl. open-question anchor. Defaults if none exists yet."""
    db = await get_db()
    cursor = await db.execute(
        "SELECT good_answers, streak, completed, understood, mastered, "
        "offene_question, offene_basis, offene_streak FROM block_progress "
        "WHERE topic = ? AND block = ?",
        (topic, block),
    )
    row = await cursor.fetchone()
    if row is None:
        return {"good_answers": 0, "streak": 0, "completed": None, "understood": None,
                "mastered": None, "offene_question": None, "offene_basis": None, "offene_streak": None}
    return {"good_answers": row[0], "streak": row[1], "completed": row[2], "understood": row[3],
            "mastered": row[4], "offene_question": row[5], "offene_basis": row[6], "offene_streak": row[7]}


async def set_open_question(topic: str, block: str, question: str, basis: int, streak: int) -> None:
    """Freeze base + streak BEFORE the now-open question (anchor for idempotent re-rating)."""
    db = await get_db()
    now = _now()
    await db.execute(
        """INSERT INTO block_progress (topic, block, offene_question, offene_basis, offene_streak, updated_at)
           VALUES (?, ?, ?, ?, ?, ?)
           ON CONFLICT(topic, block) DO UPDATE SET
               offene_question = excluded.offene_question, offene_basis = excluded.offene_basis,
               offene_streak = excluded.offene_streak, updated_at = excluded.updated_at""",
        (topic, block, question, basis, streak, now),
    )
    await db.commit()


async def set_block_score_and_streak(topic: str, block: str, score: int, streak: int) -> tuple[int, int]:
    """Set score + streak atomically (clamped by the caller). Returns (score, streak)."""
    db = await get_db()
    await db.execute(
        """INSERT INTO block_progress (topic, block, good_answers, streak, updated_at)
           VALUES (?, ?, ?, ?, ?)
           ON CONFLICT(topic, block) DO UPDATE SET
               good_answers = excluded.good_answers, streak = excluded.streak,
               updated_at = excluded.updated_at""",
        (topic, block, score, streak, _now()),
    )
    await db.commit()
    return score, streak


async def delete_block_progress(topic: str, block: str) -> None:
    """Reset the progress of ONE block: delete the row (score/streak/flags/open question).
    If the row is missing, get_block_progress returns defaults (0) — i.e. a full reset."""
    db = await get_db()
    await db.execute("DELETE FROM block_progress WHERE topic = ? AND block = ?", (topic, block))
    await db.commit()


# Sub-level from the two orthogonal columns: peripheral → 4 (V), otherwise level (learning-path position):
# beginner/NULL → 1, advanced → 2, expert → 3. Old values (einfach/mittel/schwer) are
# mapped in backward-compatibly. Drives the guide view A/F/E/V + cap (unlocked subs × 25).
_LEVEL_CASE = """CASE
    WHEN relevance = 'peripheral' THEN 4
    WHEN level IN ('advanced', 'medium') THEN 2
    WHEN level IN ('expert', 'hard') THEN 3
    ELSE 1
END"""


def _empty_levels() -> dict[int, int]:
    return {1: 0, 2: 0, 3: 0, 4: 0}


async def subs_per_level(topic: str, block: str) -> dict[int, int]:
    """Consensus subblocks of a block per level 1–4. Pass in the raw block title."""
    from textkit import _norm_title
    db = await get_db()
    cursor = await db.execute(
        # alias must NOT be named `level`: SQLite resolves an ambiguous GROUP BY name to
        # the source COLUMN, which silently miscounts peripheral subs
        f"SELECT {_LEVEL_CASE} AS lv, COUNT(*) FROM subblocks "
        "WHERE topic = ? AND block_norm = ? AND status = 'consensus' GROUP BY lv",
        (topic, _norm_title(block)),
    )
    out = _empty_levels()
    for level, n in await cursor.fetchall():
        out[level] = n
    return out


async def subs_per_level_raw(topic: str) -> dict[str, dict[int, int]]:
    """Subblocks per level, grouped by RAW block title (= guide section title)."""
    db = await get_db()
    cursor = await db.execute(
        f"SELECT block, {_LEVEL_CASE} AS lv, COUNT(*) FROM subblocks "
        "WHERE topic = ? AND status = 'consensus' GROUP BY block, lv",
        (topic,),
    )
    out: dict[str, dict[int, int]] = {}
    for b, level, n in await cursor.fetchall():
        out.setdefault(b, _empty_levels())[level] = n
    return out


async def subs_per_level_all() -> dict[tuple[str, str], dict[int, int]]:
    """Subblocks per level per (topic, block_norm) — for the topic-wide levels derivation."""
    db = await get_db()
    cursor = await db.execute(
        f"SELECT topic, block_norm, {_LEVEL_CASE} AS lv, COUNT(*) FROM subblocks "
        "WHERE status = 'consensus' GROUP BY topic, block_norm, lv"
    )
    out: dict[tuple[str, str], dict[int, int]] = {}
    for t, bn, level, n in await cursor.fetchall():
        out.setdefault((t, bn), _empty_levels())[level] = n
    return out


async def subs_with_level(topic: str, block: str) -> list[dict]:
    """Consensus subblocks of a block with title, sub_norm and level 1–4."""
    from textkit import _norm_title
    db = await get_db()
    cursor = await db.execute(
        f"SELECT sub_title, sub_norm, {_LEVEL_CASE} AS level FROM subblocks "
        "WHERE topic = ? AND block_norm = ? AND status = 'consensus'",
        (topic, _norm_title(block)),
    )
    return [{"title": t, "norm": sn, "level": e} for t, sn, e in await cursor.fetchall()]


async def list_block_scores_all() -> list[tuple[str, str, int]]:
    """(topic, block, good_answers) per block — raw data for the levels derivation."""
    db = await get_db()
    cursor = await db.execute("SELECT topic, block, good_answers FROM block_progress")
    return [(t, b, n) for t, b, n in await cursor.fetchall()]


async def delete_block_data(topic: str) -> None:
    db = await get_db()
    await db.execute("DELETE FROM block_progress WHERE topic = ?", (topic,))
    await db.commit()


# --- Blocks pipeline content: inventory / subblocks / question pattern / coverage / state / source ---

async def upsert_block(topic: str, title_norm: str, title: str, description: str = "",
                          sources: list | None = None, reader: str | None = None) -> None:
    """Insert a candidate or union the reader set. The first description is kept.

    `reader` = ID of the research reader (e.g. "a5-1"). The union runs race-free in
    ONE statement (json1), because several reader coroutines upsert concurrently — a
    read-modify-write across `await` would lose members. `mentions` stays in sync
    with `len(reader)`. `reader=None` (e.g. from `_set_inventar`) leaves the set unchanged."""
    db = await get_db()
    rid = reader if isinstance(reader, str) and reader else None
    await db.execute(
        """INSERT INTO blocks (topic, title_norm, title, description, mentions, status, sources, reader, updated_at)
           VALUES (?, ?, ?, ?, 1, 'candidate', ?, ?, ?)
           ON CONFLICT(topic, title_norm) DO UPDATE SET
               reader = (SELECT json_group_array(v) FROM (
                            SELECT value AS v FROM json_each(blocks.reader)
                            UNION SELECT ? WHERE ? IS NOT NULL)),
               mentions = (SELECT count(*) FROM (
                            SELECT value AS v FROM json_each(blocks.reader)
                            UNION SELECT ? WHERE ? IS NOT NULL)),
               sources = excluded.sources, updated_at = excluded.updated_at""",
        (topic, title_norm, title, description,
         json.dumps(sources or [], ensure_ascii=False),
         json.dumps([rid] if rid else [], ensure_ascii=False), _now(),
         rid, rid, rid, rid),
    )
    await db.commit()


async def list_blocks(topic: str, status: str | None = None) -> list[dict]:
    db = await get_db()
    if status is None:
        cursor = await db.execute("SELECT * FROM blocks WHERE topic = ? ORDER BY rowid", (topic,))
    else:
        cursor = await db.execute("SELECT * FROM blocks WHERE topic = ? AND status = ? ORDER BY rowid", (topic, status))
    rows = await cursor.fetchall()
    out = []
    for row in rows:
        d = _row_to_dict(row, cursor)
        d["sources"] = json.loads(d.get("sources") or "[]")
        d["reader"] = json.loads(d.get("reader") or "[]")
        out.append(d)
    return out


async def set_block_status(topic: str, title_norm: str, status: str, title: str | None = None, description: str | None = None, neu_norm: str | None = None) -> None:
    """Set status; optionally update title/description (e.g. after a semantic merge).
    `neu_norm` renames the norm key (clarification: reference title → meaningful name). Only
    safe while no subblocks/facts are attached to the old `title_norm` yet."""
    fields = {"status": status, "updated_at": _now()}
    if title is not None:
        fields["title"] = title
    if description is not None:
        fields["description"] = description
    if neu_norm is not None and neu_norm != title_norm:
        fields["title_norm"] = neu_norm
    await _update("blocks", fields, {"topic": topic, "title_norm": title_norm})


async def delete_blocks(topic: str) -> None:
    db = await get_db()
    await db.execute("DELETE FROM blocks WHERE topic = ?", (topic,))
    await db.commit()


# ── Kanban dataflow (generic card layer, boards 'inventory' + 'artefacts') ────────
def _now_plus(seconds: float) -> str:
    from datetime import datetime, timedelta, timezone
    return (datetime.now(timezone.utc) + timedelta(seconds=seconds)).isoformat()


def _card(row, cursor) -> dict:
    """Row → dict with the JSON payload decoded (payload keys stay under 'payload')."""
    d = _row_to_dict(row, cursor)
    try:
        d["payload"] = json.loads(d.get("payload") or "{}")
    except (TypeError, ValueError):
        d["payload"] = {}
    return d


async def kanban_pull(topic: str, board: str, stage: str, limit: int) -> list[dict]:
    """`limit` ready cards of `stage` (backoff expired). LPT: cards carrying a `subs_n`
    payload field (board 2, set after subblocks) are pulled BIGGEST first — the longest
    block starts earliest and stops dominating the makespan tail. `n_size` (board 1,
    reader count) is the coarser fallback estimate. Others stay FIFO."""
    db = await get_db()
    cursor = await db.execute(
        """SELECT * FROM kanban_cards WHERE topic = ? AND board = ? AND stage = ? AND not_before <= ?
           ORDER BY COALESCE(json_extract(payload, '$.subs_n'),
                             json_extract(payload, '$.n_size'), 0) DESC, updated_at LIMIT ?""",
        (topic, board, stage, _now(), limit))
    return [_card(row, cursor) for row in await cursor.fetchall()]


async def kanban_count(topic: str, stages, board: str | None = None) -> int:
    """Cards sitting in any of `stages` (str or list) — queue length / quiescence.
    Cards in backoff still count: their work is not done. `board=None` = across boards."""
    if isinstance(stages, str):
        stages = [stages]
    if not stages:
        return 0
    db = await get_db()
    ph = ",".join("?" * len(stages))
    sql = f"SELECT count(*) FROM kanban_cards WHERE topic = ? AND stage IN ({ph})"
    args: tuple = (topic, *stages)
    if board:
        sql += " AND board = ?"
        args += (board,)
    cursor = await db.execute(sql, args)
    return (await cursor.fetchone())[0]


async def kanban_advance(topic: str, board: str, card_id: str, stage: str) -> None:
    """Move a card to `stage` (next column, or back for rework). Clears backoff/error."""
    await kanban_advance_many(topic, board, [(card_id, stage)])


async def add_event(topic: str, kind: str, key: str = "", label: str = "", status: str = "",
                    dur_ms: int | None = None, wait_ms: int | None = None, meta: dict | None = None) -> None:
    """One pipeline-history row, own commit. Callers treat this as fire-and-forget."""
    db = await get_db()
    await db.execute(
        "INSERT INTO events (topic, ts, kind, key, label, status, dur_ms, wait_ms, meta) VALUES (?,?,?,?,?,?,?,?,?)",
        (topic, _now(), kind, key, label, status, dur_ms, wait_ms,
         json.dumps(meta or {}, ensure_ascii=False)))
    await db.commit()


async def _add_events_many(db, topic: str, rows: list[tuple]) -> None:
    """Batch insert WITHOUT commit — must run inside the caller's transaction
    (kanban_advance_many) so the event batch stays atomic with the moves."""
    now = _now()
    await db.executemany(
        "INSERT INTO events (topic, ts, kind, key, label, status, dur_ms, wait_ms, meta) VALUES (?,?,?,?,?,?,?,?,?)",
        [(topic, now, kind, key, label, status, None, None, "{}")
         for kind, key, label, status in rows])


async def kanban_advance_many(topic: str, board: str, moves: list[tuple[str, str]]) -> None:
    """Batch stage moves in ONE commit (the flow advances whole packages)."""
    if not moves:
        return
    db = await get_db()
    now = _now()
    await db.executemany(
        """UPDATE kanban_cards SET stage = ?, retries = 0, not_before = '', last_error = NULL, updated_at = ?
           WHERE topic = ? AND board = ? AND card_id = ?""",
        [(stage, now, topic, board, cid) for cid, stage in moves])
    await _add_events_many(db, topic, [("stage", f"{board}:{cid}", "", stage) for cid, stage in moves])
    await db.commit()


async def kanban_upsert_card(topic: str, board: str, card_id: str, kind: str, stage: str,
                             payload: dict | None = None) -> None:
    """Insert or overwrite a card (stable ids → growing clusters upsert, never duplicate).
    payload=None keeps the existing payload on conflict."""
    db = await get_db()
    await db.execute(
        """INSERT INTO kanban_cards (topic, board, card_id, kind, stage, payload, updated_at)
           VALUES (?, ?, ?, ?, ?, COALESCE(?, '{}'), ?)
           ON CONFLICT(topic, board, card_id) DO UPDATE SET
               kind = excluded.kind, stage = excluded.stage,
               payload = COALESCE(?, kanban_cards.payload),
               retries = 0, not_before = '', last_error = NULL, updated_at = excluded.updated_at""",
        (topic, board, card_id, kind, stage,
         json.dumps(payload, ensure_ascii=False) if payload is not None else None, _now(),
         json.dumps(payload, ensure_ascii=False) if payload is not None else None))
    await db.commit()


async def kanban_set_payload(topic: str, board: str, card_id: str, payload: dict) -> None:
    db = await get_db()
    await db.execute(
        "UPDATE kanban_cards SET payload = ?, updated_at = ? WHERE topic = ? AND board = ? AND card_id = ?",
        (json.dumps(payload, ensure_ascii=False), _now(), topic, board, card_id))
    await db.commit()


async def kanban_get_card(topic: str, board: str, card_id: str) -> dict | None:
    db = await get_db()
    cursor = await db.execute(
        "SELECT * FROM kanban_cards WHERE topic = ? AND board = ? AND card_id = ?", (topic, board, card_id))
    row = await cursor.fetchone()
    return _card(row, cursor) if row else None


async def kanban_cards(topic: str, board: str | None = None, stage: str | None = None,
                       kind: str | None = None) -> list[dict]:
    db = await get_db()
    sql, args = "SELECT * FROM kanban_cards WHERE topic = ?", [topic]
    for col, val in (("board", board), ("stage", stage), ("kind", kind)):
        if val is not None:
            sql += f" AND {col} = ?"
            args.append(val)
    cursor = await db.execute(sql, args)
    return [_card(row, cursor) for row in await cursor.fetchall()]


async def kanban_fail_card(topic: str, board: str, card_id: str, error: str,
                           max_retries: int, backoff_base: float = 30.0) -> bool:
    """Register a processing failure: retries++, exponential backoff (not_before), and after
    `max_retries` → stage 'dead' (dead-letter, requeue-able). → True if the card went dead."""
    db = await get_db()
    cursor = await db.execute(
        "SELECT retries FROM kanban_cards WHERE topic = ? AND board = ? AND card_id = ?",
        (topic, board, card_id))
    row = await cursor.fetchone()
    if row is None:
        return False
    retries = (row[0] or 0) + 1
    dead = retries >= max_retries
    if dead:
        await db.execute(
            """UPDATE kanban_cards SET stage = 'dead', retries = ?, not_before = '',
               last_error = ?, updated_at = ? WHERE topic = ? AND board = ? AND card_id = ?""",
            (retries, error[:500], _now(), topic, board, card_id))
    else:
        await db.execute(
            """UPDATE kanban_cards SET retries = ?, not_before = ?, last_error = ?, updated_at = ?
               WHERE topic = ? AND board = ? AND card_id = ?""",
            (retries, _now_plus(backoff_base * (2 ** (retries - 1))), error[:500], _now(),
             topic, board, card_id))
    await db.execute(
        "INSERT INTO events (topic, ts, kind, key, label, status, dur_ms, wait_ms, meta) VALUES (?,?,?,?,?,?,?,?,?)",
        (topic, _now(), "fail", f"{board}:{card_id}", "", "dead" if dead else f"retry{retries}",
         None, None, json.dumps({"error": error[:200]}, ensure_ascii=False)))
    await db.commit()
    return dead


async def kanban_dead(topic: str) -> list[dict]:
    """Dead-letter cards across boards (for the board UI + requeue)."""
    return await kanban_cards(topic, stage="dead")


async def kanban_requeue_dead(topic: str, board: str, stage: str) -> int:
    """dead → `stage` (fresh retries). → number of requeued cards."""
    db = await get_db()
    cursor = await db.execute(
        """UPDATE kanban_cards SET stage = ?, retries = 0, not_before = '', last_error = NULL, updated_at = ?
           WHERE topic = ? AND board = ? AND stage = 'dead'""",
        (stage, _now(), topic, board))
    await _add_events_many(db, topic, [("reset", f"{board}:requeue-dead", "", stage)])
    await db.commit()
    return cursor.rowcount


async def kanban_stage_counts(topic: str) -> dict[str, dict[str, int]]:
    """{board: {stage: count}} — the live board."""
    db = await get_db()
    cursor = await db.execute(
        "SELECT board, stage, count(*) FROM kanban_cards WHERE topic = ? GROUP BY board, stage", (topic,))
    out: dict[str, dict[str, int]] = {}
    for board, stage, n in await cursor.fetchall():
        out.setdefault(board, {})[stage] = n
    return out


async def kanban_stage_cards(topic: str, board: str, stage: str, limit: int = 20) -> list[dict]:
    """Newest `limit` cards of one column (for the live card display)."""
    db = await get_db()
    cursor = await db.execute(
        """SELECT * FROM kanban_cards WHERE topic = ? AND board = ? AND stage = ?
           ORDER BY updated_at DESC LIMIT ?""",
        (topic, board, stage, limit))
    return [_card(row, cursor) for row in await cursor.fetchall()]


async def kanban_delete_cards(topic: str, board: str, kind: str | None = None) -> None:
    """Delete derived cards (board reset) — kind=None wipes the whole board."""
    db = await get_db()
    if kind:
        await db.execute("DELETE FROM kanban_cards WHERE topic = ? AND board = ? AND kind = ?",
                         (topic, board, kind))
    else:
        await db.execute("DELETE FROM kanban_cards WHERE topic = ? AND board = ?", (topic, board))
    await db.commit()


async def kanban_reset(topic: str, board: str | None = None) -> None:
    db = await get_db()
    if board:
        await db.execute("DELETE FROM kanban_cards WHERE topic = ? AND board = ?", (topic, board))
        if board == "inventory":
            await db.execute("DELETE FROM kanban_members WHERE topic = ?", (topic,))
    else:
        await db.execute("DELETE FROM kanban_cards WHERE topic = ?", (topic,))
        await db.execute("DELETE FROM kanban_members WHERE topic = ?", (topic,))
    await db.commit()


async def kanban_set_members(topic: str, group_id: str, members: list[str]) -> None:
    """Replace the member set of a cluster (one member belongs to exactly one cluster)."""
    db = await get_db()
    await db.execute("DELETE FROM kanban_members WHERE topic = ? AND group_id = ?", (topic, group_id))
    await db.executemany(
        """INSERT INTO kanban_members (topic, member_id, group_id) VALUES (?, ?, ?)
           ON CONFLICT(topic, member_id) DO UPDATE SET group_id = excluded.group_id""",
        [(topic, m, group_id) for m in members])
    await db.commit()


async def kanban_members_of(topic: str, group_id: str) -> list[str]:
    db = await get_db()
    cursor = await db.execute(
        "SELECT member_id FROM kanban_members WHERE topic = ? AND group_id = ?", (topic, group_id))
    return [r[0] for r in await cursor.fetchall()]


# ── Guide board (one card per block, linear stages) ──────────────────────────────
async def upsert_guide_card(topic: str, format: str, block_norm: str, block: str,
                            stage: str = "lernziele") -> None:
    """Insert a card; an existing one keeps its stage/progress (resume)."""
    db = await get_db()
    await db.execute(
        """INSERT INTO guide_cards (topic, format, block_norm, block, stage, updated_at)
           VALUES (?, ?, ?, ?, ?, ?)
           ON CONFLICT(topic, format, block_norm) DO UPDATE SET
               block = excluded.block, updated_at = excluded.updated_at""",
        (topic, format, block_norm, block, stage, _now()))
    await db.commit()


async def list_guide_cards(topic: str, format: str) -> list[dict]:
    db = await get_db()
    cursor = await db.execute(
        "SELECT * FROM guide_cards WHERE topic = ? AND format = ? ORDER BY ord, block_norm",
        (topic, format))
    return [_row_to_dict(row, cursor) for row in await cursor.fetchall()]


async def set_guide_card(topic: str, format: str, block_norm: str, **fields) -> None:
    if not fields:
        return
    db = await get_db()
    cols = ", ".join(f"{k} = ?" for k in fields)
    await db.execute(
        f"UPDATE guide_cards SET {cols}, updated_at = ? WHERE topic = ? AND format = ? AND block_norm = ?",
        (*fields.values(), _now(), topic, format, block_norm))
    if "stage" in fields:  # the guide board moves stages here, not via kanban_advance_many
        await _add_events_many(db, topic, [("stage", f"guide:{format}:{block_norm}", "", fields["stage"])])
    elif fields.get("status") == "error":  # guide cards fail here, not via kanban_fail_card
        await _add_events_many(db, topic, [("fail", f"guide:{format}:{block_norm}", "",
                                            str(fields.get("gate_info", ""))[:200])])
    await db.commit()


async def guide_stage_counts(topic: str, format: str) -> dict[str, int]:
    db = await get_db()
    cursor = await db.execute(
        "SELECT stage, count(*) FROM guide_cards WHERE topic = ? AND format = ? GROUP BY stage",
        (topic, format))
    return {stage: n for stage, n in await cursor.fetchall()}


async def reset_guide_cards_from_stage(topic: str, format: str, stages: list[str],
                                       to_stage: str, clear_md: bool = False) -> int:
    """Cards sitting in any of `stages` → back to `to_stage` (fresh rounds/gate info)."""
    if not stages:
        return 0
    db = await get_db()
    ph = ",".join("?" * len(stages))
    md = ", md = ''" if clear_md else ""
    cursor = await db.execute(
        f"""UPDATE guide_cards SET stage = ?, status = 'open', writer_rounds = 0,
            gate_info = ''{md}, updated_at = ?
            WHERE topic = ? AND format = ? AND stage IN ({ph})""",
        (to_stage, _now(), topic, format, *stages))
    await db.commit()
    return cursor.rowcount


async def delete_guide_board(topic: str, format: str | None = None) -> None:
    db = await get_db()
    if format:
        await db.execute("DELETE FROM guide_cards WHERE topic = ? AND format = ?", (topic, format))
        cursor = await db.execute("SELECT count(*) FROM guide_cards WHERE topic = ?", (topic,))
        if (await cursor.fetchone())[0] == 0:
            await db.execute("DELETE FROM guide_lernziele WHERE topic = ?", (topic,))
    else:
        await db.execute("DELETE FROM guide_cards WHERE topic = ?", (topic,))
        await db.execute("DELETE FROM guide_lernziele WHERE topic = ?", (topic,))
    await db.commit()


async def put_lernziel(topic: str, block_norm: str, ziel_id: str, text: str, sub_norm: str = "") -> None:
    db = await get_db()
    await db.execute(
        """INSERT INTO guide_lernziele (topic, block_norm, ziel_id, text, sub_norm, covered, updated_at)
           VALUES (?, ?, ?, ?, ?, 0, ?)
           ON CONFLICT(topic, block_norm, ziel_id) DO UPDATE SET
               text = excluded.text, sub_norm = excluded.sub_norm, updated_at = excluded.updated_at""",
        (topic, block_norm, ziel_id, text, sub_norm, _now()))
    await db.commit()


async def list_lernziele(topic: str, block_norm: str | None = None) -> list[dict]:
    db = await get_db()
    if block_norm is None:
        cursor = await db.execute(
            "SELECT * FROM guide_lernziele WHERE topic = ? ORDER BY block_norm, ziel_id", (topic,))
    else:
        cursor = await db.execute(
            "SELECT * FROM guide_lernziele WHERE topic = ? AND block_norm = ? ORDER BY ziel_id",
            (topic, block_norm))
    return [_row_to_dict(row, cursor) for row in await cursor.fetchall()]


async def set_ziel_covered(topic: str, block_norm: str, ziel_id: str, covered: bool) -> None:
    db = await get_db()
    await db.execute(
        "UPDATE guide_lernziele SET covered = ?, updated_at = ? WHERE topic = ? AND block_norm = ? AND ziel_id = ?",
        (1 if covered else 0, _now(), topic, block_norm, ziel_id))
    await db.commit()


async def delete_lernziele(topic: str, block_norm: str) -> None:
    db = await get_db()
    await db.execute("DELETE FROM guide_lernziele WHERE topic = ? AND block_norm = ?", (topic, block_norm))
    await db.commit()


async def kanban_membership(topic: str) -> dict[str, str]:
    """{member_id: group_id} for the whole topic (the cluster worker's working map)."""
    db = await get_db()
    cursor = await db.execute("SELECT member_id, group_id FROM kanban_members WHERE topic = ?", (topic,))
    return {m: g for m, g in await cursor.fetchall()}


async def kanban_set_member(topic: str, member_id: str, group_id: str) -> None:
    db = await get_db()
    await db.execute(
        """INSERT INTO kanban_members (topic, member_id, group_id) VALUES (?, ?, ?)
           ON CONFLICT(topic, member_id) DO UPDATE SET group_id = excluded.group_id""",
        (topic, member_id, group_id))
    await db.commit()


async def kanban_add_title(topic: str, board: str, card_id: str, title: str,
                           description: str, source: str, reader: str) -> bool:
    """Ingest one research title (stage 'ingest'). An exact dupe folds instead of duplicating:
    reader-set union, source union, longer description wins; stage stays untouched (a title
    already consumed into a cluster is not re-queued). NOT concurrency-safe across awaits —
    callers serialize through one ingest lock. → True if the card is new."""
    row = await kanban_get_card(topic, board, card_id)
    if row is None:
        payload = {"title": title, "description": description,
                   "sources": [source] if source else [], "readers": [reader]}
        db = await get_db()
        await db.execute(
            """INSERT INTO kanban_cards (topic, board, card_id, kind, stage, payload, updated_at)
               VALUES (?, ?, ?, 'title', 'ingest', ?, ?)""",
            (topic, board, card_id, json.dumps(payload, ensure_ascii=False), _now()))
        await db.commit()
        return True
    p = row["payload"]
    p["readers"] = list(dict.fromkeys((p.get("readers") or []) + [reader]))
    p["sources"] = list(dict.fromkeys((p.get("sources") or []) + ([source] if source else [])))
    if len(description or "") > len(p.get("description") or ""):
        p["description"] = description
    await kanban_set_payload(topic, board, card_id, p)
    return False


async def upsert_subblock(topic: str, block_norm: str, sub_norm: str, block: str, sub_title: str) -> None:
    db = await get_db()
    await db.execute(
        """INSERT INTO subblocks (topic, block_norm, sub_norm, block, sub_title, mentions, status, updated_at)
           VALUES (?, ?, ?, ?, ?, 1, 'candidate', ?)
           ON CONFLICT(topic, block_norm, sub_norm) DO UPDATE SET
               mentions = mentions + 1, updated_at = excluded.updated_at""",
        (topic, block_norm, sub_norm, block, sub_title, _now()),
    )
    await db.commit()


async def put_subblock(topic: str, block_norm: str, sub_norm: str, block: str, sub_title: str,
                          level: str | None = None, relevance: str | None = None,
                          facts: str | None = None, status: str = "consensus") -> None:
    """Insert/update WITHOUT a mention counter (mirror from the sidecar). Overwrite
    level/relevance/facts only when a new value is passed (COALESCE protects existing data)."""
    db = await get_db()
    await db.execute(
        """INSERT INTO subblocks (topic, block_norm, sub_norm, block, sub_title, mentions, level, relevance, facts, status, updated_at)
           VALUES (?, ?, ?, ?, ?, 1, ?, ?, COALESCE(?, ''), ?, ?)
           ON CONFLICT(topic, block_norm, sub_norm) DO UPDATE SET
               block = excluded.block, sub_title = excluded.sub_title,
               level = COALESCE(excluded.level, subblocks.level),
               relevance = COALESCE(excluded.relevance, subblocks.relevance),
               facts = COALESCE(NULLIF(excluded.facts, ''), subblocks.facts),
               status = excluded.status, updated_at = excluded.updated_at""",
        (topic, block_norm, sub_norm, block, sub_title, level, relevance, facts, status, _now()),
    )
    await db.commit()


async def list_subblocks(topic: str, block_norm: str | None = None) -> list[dict]:
    db = await get_db()
    if block_norm is None:
        cursor = await db.execute("SELECT * FROM subblocks WHERE topic = ? ORDER BY rowid", (topic,))
    else:
        cursor = await db.execute(
            "SELECT * FROM subblocks WHERE topic = ? AND block_norm = ? ORDER BY rowid", (topic, block_norm)
        )
    rows = await cursor.fetchall()
    return [_row_to_dict(row, cursor) for row in rows]


async def set_subblock_fields(topic: str, block_norm: str, sub_norm: str, **fields) -> None:
    """Set fields (level/relevance/status/sub_title) of a subblock row."""
    fields["updated_at"] = _now()
    await _update("subblocks", fields, {"topic": topic, "block_norm": block_norm, "sub_norm": sub_norm})


async def delete_subblocks(topic: str, block_norm: str | None = None) -> None:
    db = await get_db()
    if block_norm is None:
        await db.execute("DELETE FROM subblocks WHERE topic = ?", (topic,))
    else:
        await db.execute("DELETE FROM subblocks WHERE topic = ? AND block_norm = ?", (topic, block_norm))
    await db.commit()


async def upsert_question_pattern(topic: str, block_norm: str, sub_norm: str, block: str, sub_title: str, question: str) -> None:
    db = await get_db()
    await db.execute(
        """INSERT INTO question_pattern (topic, block_norm, sub_norm, block, sub_title, question, updated_at)
           VALUES (?, ?, ?, ?, ?, ?, ?)
           ON CONFLICT(topic, block_norm, sub_norm) DO UPDATE SET
               sub_title = excluded.sub_title, question = excluded.question, updated_at = excluded.updated_at""",
        (topic, block_norm, sub_norm, block, sub_title, question, _now()),
    )
    await db.commit()


async def list_question_pattern(topic: str, block_norm: str | None = None) -> list[dict]:
    db = await get_db()
    if block_norm is None:
        cursor = await db.execute("SELECT * FROM question_pattern WHERE topic = ? ORDER BY rowid", (topic,))
    else:
        cursor = await db.execute(
            "SELECT * FROM question_pattern WHERE topic = ? AND block_norm = ? ORDER BY rowid", (topic, block_norm)
        )
    rows = await cursor.fetchall()
    return [_row_to_dict(row, cursor) for row in rows]


async def count_question_pattern_blocks(topic: str) -> int:
    """Blocks that have at least one exam question pattern."""
    db = await get_db()
    cur = await db.execute("SELECT COUNT(DISTINCT block_norm) FROM question_pattern WHERE topic = ?", (topic,))
    return (await cur.fetchone())[0]


async def count_sub_artefakte(topic: str) -> int:
    db = await get_db()
    cur = await db.execute("SELECT COUNT(*) FROM sub_artefakte WHERE topic = ?", (topic,))
    return (await cur.fetchone())[0]


async def event_span(topic: str) -> int:
    """Minutes between first and last pipeline event of the topic (0 if none)."""
    db = await get_db()
    cur = await db.execute("SELECT MIN(ts), MAX(ts) FROM events WHERE topic = ?", (topic,))
    lo, hi = await cur.fetchone()
    if not lo or not hi:
        return 0
    from datetime import datetime
    return int((datetime.fromisoformat(hi) - datetime.fromisoformat(lo)).total_seconds() // 60)


async def delete_question_pattern(topic: str, block_norm: str | None = None) -> None:
    db = await get_db()
    if block_norm is None:
        await db.execute("DELETE FROM question_pattern WHERE topic = ?", (topic,))
    else:
        await db.execute("DELETE FROM question_pattern WHERE topic = ? AND block_norm = ?", (topic, block_norm))
    await db.commit()


async def mark_sources_read_done(topic: str, sources: list[str]) -> None:
    """Mark the cited crawl pages as read_done (research-loop coverage)."""
    if not sources:
        return
    db = await get_db()
    now = _now()
    await db.executemany(
        """INSERT INTO research_coverage (topic, source, read_done, updated_at) VALUES (?, ?, 1, ?)
           ON CONFLICT(topic, source) DO UPDATE SET read_done = 1, updated_at = excluded.updated_at""",
        [(topic, q, now) for q in sources],
    )
    await db.commit()


async def mark_content(topic: str, content: list[str], noise: list[str]) -> None:
    """Store the triage result per crawl page: content=1 (content) or 0 (noise)."""
    db = await get_db()
    now = _now()
    rows = [(topic, q, 1, now) for q in content] + [(topic, q, 0, now) for q in noise]
    if not rows:
        return
    await db.executemany(
        """INSERT INTO research_coverage (topic, source, content, updated_at) VALUES (?, ?, ?, ?)
           ON CONFLICT(topic, source) DO UPDATE SET content = excluded.content, updated_at = excluded.updated_at""",
        rows,
    )
    await db.commit()


async def list_content(topic: str) -> list[str]:
    """Crawl pages the triage marked as content (content=1)."""
    db = await get_db()
    cursor = await db.execute(
        "SELECT source FROM research_coverage WHERE topic = ? AND content = 1 ORDER BY source", (topic,)
    )
    return [r[0] for r in await cursor.fetchall()]


async def delete_coverage(topic: str) -> None:
    db = await get_db()
    await db.execute("DELETE FROM research_coverage WHERE topic = ?", (topic,))
    await db.commit()


async def set_step_status(topic: str, step: str, status: str) -> None:
    db = await get_db()
    await db.execute(
        """INSERT INTO pipeline_state (topic, step, status, updated_at) VALUES (?, ?, ?, ?)
           ON CONFLICT(topic, step) DO UPDATE SET status = excluded.status, updated_at = excluded.updated_at""",
        (topic, step, status, _now()),
    )
    await db.commit()


async def get_step_status(topic: str, step: str) -> str:
    db = await get_db()
    cursor = await db.execute(
        "SELECT status FROM pipeline_state WHERE topic = ? AND step = ?", (topic, step)
    )
    row = await cursor.fetchone()
    return row[0] if row else "open"








async def delete_source(topic: str) -> None:
    db = await get_db()
    await db.execute("DELETE FROM source WHERE topic = ?", (topic,))
    await db.commit()


async def set_guide_content(topic: str, format: str, content_json: str) -> None:
    """Store finished guide content (JSON blob) per topic+format."""
    db = await get_db()
    await db.execute(
        """INSERT INTO guide_content (topic, format, json, updated_at) VALUES (?, ?, ?, ?)
           ON CONFLICT(topic, format) DO UPDATE SET json = excluded.json, updated_at = excluded.updated_at""",
        (topic, format, content_json, _now()),
    )
    await db.commit()


async def get_guide_content(topic: str, format: str) -> str | None:
    db = await get_db()
    cursor = await db.execute("SELECT json FROM guide_content WHERE topic = ? AND format = ?", (topic, format))
    row = await cursor.fetchone()
    return row[0] if row else None


async def delete_guide_content(topic: str, format: str | None = None) -> None:
    db = await get_db()
    if format is None:
        await db.execute("DELETE FROM guide_content WHERE topic = ?", (topic,))
    else:
        await db.execute("DELETE FROM guide_content WHERE topic = ? AND format = ?", (topic, format))
    await db.commit()


async def set_outline(topic: str, outline_json: str) -> None:
    """Store the outline (chapter→numbers, JSON) per topic — blocks artifact for the guide."""
    db = await get_db()
    await db.execute(
        """INSERT INTO guide_outline (topic, json, updated_at) VALUES (?, ?, ?)
           ON CONFLICT(topic) DO UPDATE SET json = excluded.json, updated_at = excluded.updated_at""",
        (topic, outline_json, _now()),
    )
    await db.commit()


async def get_outline(topic: str) -> str | None:
    db = await get_db()
    cursor = await db.execute("SELECT json FROM guide_outline WHERE topic = ?", (topic,))
    row = await cursor.fetchone()
    return row[0] if row else None


async def delete_outline(topic: str) -> None:
    db = await get_db()
    await db.execute("DELETE FROM guide_outline WHERE topic = ?", (topic,))
    await db.commit()


async def put_sub_artifact(topic: str, block_norm: str, sub_norm: str, type: str,
                           data: str, block: str = "", sub_title: str = "") -> None:
    """Store one learning artifact (flashcard/example) as JSON in `data`."""
    db = await get_db()
    await db.execute(
        """INSERT INTO sub_artefakte (topic, block_norm, sub_norm, type, block, sub_title, data, updated_at)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?)
           ON CONFLICT(topic, block_norm, sub_norm, type) DO UPDATE SET
               block = excluded.block, sub_title = excluded.sub_title,
               data = excluded.data, updated_at = excluded.updated_at""",
        (topic, block_norm, sub_norm, type, block, sub_title, data, _now()),
    )
    await db.commit()


async def get_sub_artefakte(topic: str, type: str | None = None,
                            block_norm: str | None = None) -> list[dict]:
    db = await get_db()
    sql = "SELECT * FROM sub_artefakte WHERE topic = ?"
    args: list = [topic]
    if type is not None:
        sql += " AND type = ?"
        args.append(type)
    if block_norm is not None:
        sql += " AND block_norm = ?"
        args.append(block_norm)
    cursor = await db.execute(sql + " ORDER BY rowid", args)
    rows = await cursor.fetchall()
    return [_row_to_dict(row, cursor) for row in rows]


async def get_practice_progress(topic: str) -> list[dict]:
    db = await get_db()
    cursor = await db.execute(
        "SELECT block_norm, sub_norm, box, due_at FROM practice_progress WHERE topic = ?", (topic,))
    rows = await cursor.fetchall()
    return [_row_to_dict(row, cursor) for row in rows]


async def upsert_practice_progress(topic: str, block_norm: str, sub_norm: str,
                                   box: int, due_at: str) -> None:
    db = await get_db()
    await db.execute(
        """INSERT INTO practice_progress (topic, block_norm, sub_norm, box, due_at, updated_at)
           VALUES (?, ?, ?, ?, ?, ?)
           ON CONFLICT(topic, block_norm, sub_norm)
           DO UPDATE SET box = excluded.box, due_at = excluded.due_at,
                         updated_at = excluded.updated_at""",
        (topic, block_norm, sub_norm, box, due_at, _now()))
    await db.commit()


async def sub_levels_norm(topic: str) -> dict[tuple[str, str], int]:
    """(block_norm, sub_norm) → level 1-4 for the consensus subs — the practice deck's gate."""
    db = await get_db()
    cursor = await db.execute(
        f"SELECT block_norm, sub_norm, {_LEVEL_CASE} AS level FROM subblocks "
        "WHERE topic = ? AND status = 'consensus'", (topic,))
    return {(bn, sn): lv for bn, sn, lv in await cursor.fetchall()}


async def subs_per_level_norm(topic: str) -> dict[str, dict[int, int]]:
    """Subblocks per level, grouped by block_norm (sub_artefakte is norm-keyed)."""
    db = await get_db()
    cursor = await db.execute(
        f"SELECT block_norm, {_LEVEL_CASE} AS lv, COUNT(*) FROM subblocks "
        "WHERE topic = ? AND status = 'consensus' GROUP BY block_norm, lv", (topic,))
    out: dict[str, dict[int, int]] = {}
    for bn, level, n in await cursor.fetchall():
        out.setdefault(bn, _empty_levels())[level] = n
    return out


async def delete_sub_artefakte(topic: str, block_norm: str | None = None) -> None:
    db = await get_db()
    if block_norm is None:
        await db.execute("DELETE FROM sub_artefakte WHERE topic = ?", (topic,))
    else:
        await db.execute("DELETE FROM sub_artefakte WHERE topic = ? AND block_norm = ?", (topic, block_norm))
    await db.commit()


async def get_block_hurdles(topic: str, block_norm: str) -> list[str]:
    """Typical misconceptions (hurdles) of all subs of a block — as a distractor pool for the quiz.
    Reads from subblocks.facts (JSON blob); empty/missing ones are skipped."""
    db = await get_db()
    cursor = await db.execute(
        "SELECT facts FROM subblocks WHERE topic = ? AND block_norm = ?", (topic, block_norm)
    )
    rows = await cursor.fetchall()
    hurdles = []
    for (facts,) in rows:
        if not facts:
            continue
        try:
            fk = json.loads(facts)
        except (ValueError, TypeError):
            continue
        h = (fk.get("hurdles") or "").strip() if isinstance(fk, dict) else ""
        if h:
            hurdles.append(h)
    return hurdles


async def delete_topic_pipeline(topic: str) -> None:
    """Discard the blocks area of a topic (inventory/subs/pattern/coverage/state/artifacts).
    NOT the topic config `source` — that is managed separately (delete_source)."""
    db = await get_db()
    for tab in ("blocks", "subblocks", "question_pattern", "research_coverage",
                "pipeline_state", "guide_outline", "sub_artefakte", "practice_progress", "events"):
        await db.execute(f"DELETE FROM {tab} WHERE topic = ?", (topic,))
    await db.commit()