cali_scraper/scraper/relational.py

"""Relational core for spec §2.5 — bills ↔ committees ↔ members.

The rest of the pipeline persists a FLAT table (``daily_file_records`` written by
``scraper.io.write_sqlite``). QA and the downstream UI read that flat contract,
so this lane NEVER mutates it. Instead it ADDS a normalised relational schema
and a flat-preserving VIEW on top of the same SQLite database, and derives the
relations (bill↔committee, committee↔member) from the flat rows + an optional
committee-roster source.

FROZEN CONTRACTS honoured (see task t_2736f40d / parent t_21cd58a9):

* ``daily_file_records`` stays fully intact (CREATE IF NOT EXISTS only).
* New tables: ``bills``, ``committees``, ``members``, ``committee_members``
  (committee↔member join), ``bill_committees`` (bill↔committee join), plus a
  ``schema_migrations`` ledger and a ``v_daily_file_records`` view.
* FOREIGN KEYS are enabled on every connection; UNIQUE constraints on the
  natural keys (bill_id+session+chamber; committee name+session+chamber;
  committee+member).
* Migration and correlation are IDEMPOTENT: CREATE IF NOT EXISTS +
  INSERT OR IGNORE on deterministic keys. Re-running adds no rows / no errors.
* NEVER FABRICATE MEMBERS. ``committee_members`` is populated ONLY from a
  supplied roster source. When a committee seen in the daily file has no roster
  entry, it is still created (with its ``session``) but is left with an EMPTY
  member set and a WARN is logged listing which committees are missing rosters —
  we record the gap, we do not invent members.

This module uses only the stdlib (sqlite3 + a tiny JSON roster reader). No new
third-party dependency is introduced.
"""

from __future__ import annotations

import json
import logging
import os
import re
import sqlite3
from typing import Any, Dict, Iterable, List, Optional

log = logging.getLogger("scraper.relate")

# ---------------------------------------------------------------------------
# Key constants
# ---------------------------------------------------------------------------

#: Default database path (spec: db/daily_file.sqlite), resolved relative to the
#: repository root so it is stable regardless of cwd.
_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DEFAULT_DB = os.path.join(_REPO_ROOT, "db", "daily_file.sqlite")

FLAT_TABLE = "daily_file_records"

#: Schema version stamped into ``schema_migrations`` by :func:`migrate`.
# v1: bills/committees/members/committee_members/bill_committees + flat view.
# v2: adds scheduled_events (the chamber "upcoming events" forward source — the
#    only genuine future-dated signal; see design/API_CONTRACT.md §5 + the
#    "what's next" work). Kept OUT of the frozen flat daily_file_records view.
SCHEMA_VERSION = 2

#: The flat columns QA + downstream rely on. Imported so the flat-preserving
#: view is byte-for-byte aligned with the contract the rest of the pipeline
#: writes via ``scraper.io.write_sqlite``.
try:    # pragma: no cover - import guard (io is a sibling in the same package)
    from .io import _COLUMNS as FLAT_COLUMNS
except Exception:    # pragma: no cover
    # Fallback if io ever changes; keeps the module importable for unit tests.
    FLAT_COLUMNS = [
        ("id", "TEXT"), ("date", "TEXT"), ("house", "TEXT"), ("source", "TEXT"),
        ("record_type", "TEXT"), ("bill_number", "TEXT"), ("bill_id", "TEXT"),
        ("session", "TEXT"), ("chamber_origin", "TEXT"), ("author", "TEXT"),
        ("short_title", "TEXT"), ("description", "TEXT"), ("committee", "TEXT"),
        ("hearing_time", "TEXT"), ("location", "TEXT"), ("item_number", "INTEGER"),
        ("supplement", "INTEGER"), ("ics_url", "TEXT"), ("agenda_id", "TEXT"),
        ("raw_selector", "TEXT"),
    ]

_FLAT_COL_ORDER = [c for c, _ in FLAT_COLUMNS]
_FLAT_COL_DEFS = ", ".join(f'"{c}" {t}' for c, t in FLAT_COLUMNS)

_SCHEMA_TABLES = (
    "bills", "committees", "members", "committee_members", "bill_committees",
    "schema_migrations",
)


# ---------------------------------------------------------------------------
# Normalisation helpers
# ---------------------------------------------------------------------------

_SLUG_RE = re.compile(r"[^a-z0-9]+")


def normalise_name(raw: Optional[str]) -> str:
    """Title-case + trim a committee/author/member display name."""
    if not raw:
        return ""
    s = re.sub(r"\s+", " ", raw.strip())
    return s


def slug(raw: Optional[str]) -> str:
    """Deterministic match key: lowercased, non-alphanumerics -> single '-'.

    Mirrors spec §6.1 ('Normalise by title-casing and stripping
    non-alphanumerics') so a roster's committee name resolves to the same
    committee the daily file references.
    """
    if not raw:
        return ""
    s = _SLUG_RE.sub("-", raw.strip().lower()).strip("-")
    return s or "unknown"


def session_key(raw: Optional[str]) -> str:
    """Coalesce a NULL/empty session to a stable sentinel ('') so the
    natural-key UNIQUE constraints stay effective and re-runs idempotent."""
    return (raw or "").strip()


def chamber_key(raw: Optional[str]) -> str:
    """Normalise chamber origin/house to a stable lowercase key."""
    return (raw or "").strip().lower() or "unknown"


# ---------------------------------------------------------------------------
# Connection + schema DDL
# ---------------------------------------------------------------------------

def connect(path: str, foreign_keys: bool = True) -> sqlite3.Connection:
    """Open the DB, enforcing FK integrity (per-connection pragma)."""
    os.makedirs(os.path.dirname(os.path.abspath(path)) or ".", exist_ok=True)
    conn = sqlite3.connect(path)
    conn.row_factory = sqlite3.Row
    if foreign_keys:
        # Must be set on the connection *before* any DDL/DML.
        conn.execute("PRAGMA foreign_keys = ON")
    return conn


def _ensure_flat_table(conn: sqlite3.Connection) -> None:
    """Create the flat contract table ONLY if missing. Never alters it."""
    conn.execute(
        f'CREATE TABLE IF NOT EXISTS {FLAT_TABLE} ('
        f"{_FLAT_COL_DEFS})"
    )


def _create_schema(conn: sqlite3.Connection) -> None:
    """Create all normalised tables + the flat-preserving view idempotently.

    Every object uses CREATE ... IF NOT EXISTS, so this is safe to run many
    times. Foreign keys link the relation tables to their parents.
    """
    conn.executescript(
        """
        CREATE TABLE IF NOT EXISTS bills (
            id             INTEGER PRIMARY KEY,
            bill_id        TEXT NOT NULL,
            bill_number    TEXT,
            session        TEXT NOT NULL DEFAULT '',
            chamber_origin TEXT NOT NULL DEFAULT 'unknown',
            house          TEXT,
            author         TEXT,
            short_title    TEXT,
            first_seen     TEXT,
            source         TEXT,
            UNIQUE (bill_id, session, chamber_origin)
        );

        CREATE TABLE IF NOT EXISTS committees (
            id           INTEGER PRIMARY KEY,
            name_key     TEXT NOT NULL,
            name_raw     TEXT NOT NULL,
            session      TEXT NOT NULL DEFAULT '',
            chamber      TEXT NOT NULL DEFAULT 'unknown',
            hearing_time TEXT,
            location     TEXT,
            source       TEXT,
            UNIQUE (name_key, session, chamber)
        );

        CREATE TABLE IF NOT EXISTS members (
            id           INTEGER PRIMARY KEY,
            name_norm    TEXT NOT NULL,
            full_name    TEXT,
            chamber      TEXT NOT NULL DEFAULT 'unknown',
            kind         TEXT NOT NULL DEFAULT 'roster',
            source       TEXT,
            UNIQUE (name_norm, chamber, kind)
        );

        CREATE TABLE IF NOT EXISTS committee_members (
            id          INTEGER PRIMARY KEY,
            committee   TEXT NOT NULL,
            member      TEXT NOT NULL,
            session     TEXT NOT NULL DEFAULT '',
            chamber     TEXT NOT NULL DEFAULT 'unknown',
            role        TEXT,
            source      TEXT,
            UNIQUE (committee, member, session, chamber)
        );

        CREATE TABLE IF NOT EXISTS bill_committees (
            id        INTEGER PRIMARY KEY,
            bill_id   TEXT NOT NULL,
            committee TEXT NOT NULL,
            session   TEXT NOT NULL DEFAULT '',
            chamber   TEXT NOT NULL DEFAULT 'unknown',
            source    TEXT,
            UNIQUE (bill_id, committee, session, chamber)
        );

        CREATE TABLE IF NOT EXISTS schema_migrations (
            version     INTEGER NOT NULL,
            applied_at  TEXT,
            note        TEXT,
            PRIMARY KEY (version)
         );

        -- v2: the chamber "upcoming events" forward source. This is the ONLY
        -- genuine *future-dated* signal (the daily file is a backward snapshot;
        -- leginfo bill status/history is backward-only). It holds the next
        -- scheduled floor/committee events parsed from the chamber events
        -- calendar (assembly.ca.gov/events, senate.ca.gov/calendar). It is
        -- deliberately OUT of the frozen flat daily_file_records view.
        CREATE TABLE IF NOT EXISTS scheduled_events (
            id          INTEGER PRIMARY KEY,
            date        TEXT NOT NULL,            -- ISO YYYY-MM-DD (event day)
            chamber     TEXT NOT NULL,            -- ASSEMBLY | SENATE
            kind        TEXT,                     -- FLOOR | COMMITTEE | OTHER
            title       TEXT NOT NULL,            -- human event title
            source_url  TEXT,                     -- where the event came from
            fetched_at  TEXT,                     -- when this page was read
            source      TEXT,                     -- assembly_events | senate_events
            UNIQUE (date, chamber, kind, title, source)
         );
      """
      )
    # Flat-preserving view: a drop-in mirror of the table QA/UI read. Column
    # order and row set are identical to daily_file_records, ordered
    # deterministically so the "flat contract unchanged" test is stable.
    # (Rebuild every migrate so the view tracks the current flat schema.)
    conn.execute(f"DROP VIEW IF EXISTS v_{FLAT_TABLE}")
    cols = ", ".join(f'"{c}"' for c in _FLAT_COL_ORDER)
    conn.execute(
        f"CREATE VIEW v_{FLAT_TABLE} AS SELECT {cols} FROM {FLAT_TABLE} "
        "ORDER BY date, house, bill_id, committee, id"
    )
    conn.commit()


def _stamp_migration(conn: sqlite3.Connection, version: int = SCHEMA_VERSION) -> None:
    conn.execute(
         "INSERT OR IGNORE INTO schema_migrations (version, applied_at, note) "
          "VALUES (?, ?, ?)",
         (version,
          __import__("datetime").date.today().isoformat(),
          "relational schema: bills/committees/members/committee_members/"
             "bill_committees + flat view + scheduled_events (v2)"),
       )
    conn.commit()


# ---------------------------------------------------------------------------
# migrate / correlate
# ---------------------------------------------------------------------------

def migrate(path: str = DEFAULT_DB, flat_table: str = FLAT_TABLE,
            drop_flat: bool = False) -> Dict[str, Any]:
    """Create the relational schema idempotently, preserving the flat table.

    Safe on a fresh DB (creates everything) and on an existing DB (no-op for
    objects that already exist). Returns a small summary dict.
    """
    conn = connect(path)
    try:
        _ensure_flat_table(conn)
        _create_schema(conn)
        _stamp_migration(conn)
        n_flat = conn.execute(f"SELECT COUNT(*) FROM {FLAT_TABLE}").fetchone()[0]
        return {"db": path, "migrated": True, "flat_rows": n_flat,
                "version": SCHEMA_VERSION,
                "objects": _list_objects(conn)}
    finally:
        conn.close()


def _list_objects(conn: sqlite3.Connection) -> List[str]:
    rows = conn.execute(
        "SELECT name FROM sqlite_master WHERE type IN ('table','view') "
        "ORDER BY type DESC, name").fetchall()
    return [r["name"] for r in rows]


def _load_flat_rows(conn: sqlite3.Connection) -> Iterable[sqlite3.Row]:
    return conn.execute(f"SELECT * FROM {FLAT_TABLE}").fetchall()


def load_rosters(path) -> Dict:
    """Load committee rosters from a JSON file (the "create new info" path).

    Format (a list of committee blocks)::

        [
          {"committee": "Budget", "session": "2025-2026", "chamber": "assembly",
           "members": [
               {"name": "Roxas", "role": "Chair"},
               {"name": "Liu", "role": "Vice Chair"}
           ]}
        ]

    Returns a dict keyed by ``(slug, session, chamber)`` -> block, so a
    committee that appears in distinct (session, chamber) slots keeps separate
    memberships instead of clobbering. Missing/empty file -> empty dict (caller
    then logs a WARN for the gap). A malformed file raises ValueError so the CLI
    can report it cleanly rather than silently producing no members.
     """
    out: Dict[tuple, Dict[str, Any]] = {}
    if not path:
        return out
    with open(path, encoding="utf-8") as fh:
        data = json.load(fh)
    blocks = data if isinstance(data, list) else data.get("committees", [])
    for blk in blocks:
        name = blk.get("committee") or blk.get("name")
        if not name:
            continue
        key = (slug(name), session_key(blk.get("session")),
               chamber_key(blk.get("chamber")))
        out[key] = {
            "name": normalise_name(name),
            "session": session_key(blk.get("session")),
            "chamber": chamber_key(blk.get("chamber")),
            "members": [
                 {"name": normalise_name(m.get("name")),
                  "role": m.get("role")}
                for m in (blk.get("members") or []) if m.get("name")
             ],
         }
    return out


def _ingest_relations(conn: sqlite3.Connection, rosters: Dict[str, Any]) -> Dict[str, Any]:
    """Derive + backfill normalised relations from the flat table.

    * ``bills``               <- flat rows with a bill_id
    * ``committees``          <- flat rows with a committee (hearing / committee
                                 tables) AND committee-like authors ('Committee
                                 on ...' strings name a committee too)
    * ``bill_committees``     <- flat rows that carry BOTH a bill_id and a
                                 committee (the "reviewed by" relation)
    * ``members``             <- roster members (kind='roster') plus bill authors
                                 (kind='author-seed') — real names, not guesses
    * ``committee_members``   <- roster membership ONLY (never fabricated)

    Returns a counts summary + the list of committees missing a roster.
    """
    counts = {"bills": 0, "committees": 0, "bill_committees": 0,
               "members": 0, "committee_members": 0,
               "committees_missing_roster": 0}
     # Helper: run an INSERT OR IGNORE and report whether it changed a row
     # (rowcount == 0 means the natural key already existed -> idempotent dedup).
    def ins_added(cur) -> int:
        return 1 if cur and cur.rowcount else 0

    seen_committees = set()    # slugs we saw in the daily file
    for row in _load_flat_rows(conn):
        bill_id = row["bill_id"] if "bill_id" in row.keys() else None
        committee = row["committee"] if "committee" in row.keys() else None
        author = row["author"] if "author" in row.keys() else None
        sess = session_key(row["session"] if "session" in row.keys() else None)
        house = row["house"] if "house" in row.keys() else None
        origin = row["chamber_origin"] if "chamber_origin" in row.keys() else None
        # chamber_origin is the origin house of a bill; hearing rows have no
        # bill so chamber_origin is None -> fall back to the publishing `house`
        # (e.g. 'assembly') so the committee slot joins to its roster cleanly.
        cham = chamber_key(origin) if origin else chamber_key(house)
        seen = "daily_file"

        # --- bills -------------------------------------------------------
        if bill_id:
            if ins_added(conn.execute(
                 "INSERT OR IGNORE INTO bills "
                 "(bill_id, bill_number, session, chamber_origin, house, "
                 " author, short_title, first_seen, source) "
                 "VALUES (?,?,?,?,?,?,?,?,?)",
                 (bill_id, row["bill_number"], sess, cham, house,
                 author, row["short_title"],
                 row["first_seen"] if "first_seen" in row.keys()
                 else None, seen))):
                counts["bills"] += 1

        # --- committee-like author names a committee ---------------------
        author_committee = None
        if author and author.strip().lower().startswith("committee"):
            # "Committee on Budget and Fiscal Review" -> "Budget and Fiscal Review"
            author_committee = re.sub(r"^committees?\s+on\s+", "",
                                      author.strip(), flags=re.I) or author.strip()
            cham = chamber_key(row["chamber_origin"] if "chamber_origin"
                              in row.keys() else None)
        for cname in {c for c in (committee, author_committee) if c}:
            ck = slug(cname)
            seen_committees.add((ck, sess, cham))
            if ins_added(conn.execute(
                 "INSERT OR IGNORE INTO committees "
                 "(name_key, name_raw, session, chamber, hearing_time, "
                 " location, source) VALUES (?,?,?,?,?,?,?)",
                 (ck, normalise_name(cname), sess, cham,
                 row["hearing_time"] if "hearing_time" in row.keys()
                 and committee else None,
                 row["location"] if "location" in row.keys()
                 and committee else None, seen))):
                counts["committees"] += 1

        # --- bill <-> committee ("reviewed by") --------------------------
        if bill_id and committee:
            if ins_added(conn.execute(
                 "INSERT OR IGNORE INTO bill_committees "
                 "(bill_id, committee, session, chamber, source) "
                 "VALUES (?,?,?,?,?)",
                 (bill_id, slug(committee), sess, cham, seen))):
                counts["bill_committees"] += 1

        # --- author -> member seed (a real legislator, not a guess) -----
        if author and not author.strip().lower().startswith("committee"):
            stem = re.sub(r"\s+et al\.$", "", author.strip())
            if re.fullmatch(r"[\w'\-]+", stem):
                if ins_added(conn.execute(
                     "INSERT OR IGNORE INTO members "
                     "(name_norm, full_name, chamber, kind, source) "
                     "VALUES (?,?,?,?,?)",
                     (slug(stem), stem, cham, "author-seed", seen))):
                    counts["members"] += 1

    # --- roster-driven committee_members (NEVER fabricated) -------------
    # `rosters` is keyed by (slug, session, chamber); populate members for each
    # roster's own slot. A roster slot may use a (session, chamber) chosen by the
    # roster source; it is not required to mirror every seen slot exactly.
    roster_slugs = {k[0] for k in rosters}
    for blk in rosters.values():
        for m in blk["members"]:
            if not m["name"]:
                continue
            if ins_added(conn.execute(
                  "INSERT OR IGNORE INTO members "
                   "(name_norm, full_name, chamber, kind, source) "
                   "VALUES (?,?,?,?,?)",
                   (slug(m["name"]), m["name"], blk["chamber"], "roster",
                    "roster"))):
                counts["members"] += 1
            if ins_added(conn.execute(
                   "INSERT OR IGNORE INTO committee_members "
                   "(committee, member, session, chamber, role, source) "
                   "VALUES (?,?,?,?,?,?)",
                   (slug(blk["name"]), slug(m["name"]), blk["session"],
                    blk["chamber"], m.get("role"), "roster"))):
                counts["committee_members"] += 1

    # --- gap reporting: committees seen but with NO roster (slug level) --
    # A seen committee slot is "missing a roster" when no roster block covers
    # that slug at all -- we still created the committee row, but its member
    # set is left empty. Per the "create new info / never fabricate" rule we
    # record the gap (+WARN) instead of inventing members.
    missing = set()
    for (ck, sess, cham) in seen_committees:
        if ck not in roster_slugs:
            missing.add((ck, sess, cham))
    counts["committees_missing_roster"] = len(missing)
    for (ck, sess, cham) in sorted(missing):
        log.warning(
              "committee %r (session=%s chamber=%s) has no roster source; "
              "membership left empty (not fabricated). Provide --roster to "
              "backfill.", ck, sess or "-", cham)

    conn.commit()
    return counts


def correlate(path: str = DEFAULT_DB, rosters: Optional[str] = None,
              rebuild: bool = False, quiet: bool = False) -> Dict[str, Any]:
    """Derive relations from the flat table into the normalised tables.

    Idempotent by default (INSERT OR IGNORE on deterministic natural keys).
    ``rebuild=True`` clears the derived join/parent tables first so a run
    reflects the current flat snapshot exactly.
    """
    conn = connect(path)
    try:
        _ensure_flat_table(conn)
        _create_schema(conn)
        _stamp_migration(conn)
        roster_data = load_rosters(rosters) if rosters else {}

        if rebuild:
            for t in ("bill_committees", "committee_members", "members",
                      "committees", "bills"):
                conn.execute(f"DELETE FROM {t}")
            conn.commit()

        counts = _ingest_relations(conn, roster_data)
        # Final absolute counts (post-dedup) for reporting.
        for k in ("bills", "committees", "bill_committees", "members",
                  "committee_members"):
            counts[k] = conn.execute(f"SELECT COUNT(*) FROM {k}").fetchone()[0]
        if not quiet:
            log.info("correlate: %s", json.dumps(counts, sort_keys=True))
        return counts
    finally:
        conn.close()


# ---------------------------------------------------------------------------
# Demonstration / verification helpers
# ---------------------------------------------------------------------------

def sample_selects(path: str = DEFAULT_DB, limit: int = 5) -> Dict[str, Any]:
    """Pull a few human-readable sample SELECTs for the completion report."""
    conn = connect(path)
    try:
        def q(sql: str, args: tuple = ()):
            return [dict(r) for r in conn.execute(sql, args).fetchall()]
        return {
            "counts": {t: conn.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0]
                       for t in ("bills", "committees", "members",
                                 "committee_members", "bill_committees")},
            "sample_bill_committees": q(
                "SELECT bc.bill_id, c.name_raw AS committee, bc.chamber, "
                "bc.session FROM bill_committees bc "
                "JOIN committees c "
                "  ON c.name_key = bc.committee "
                " AND c.chamber = bc.chamber "
                " AND c.session = bc.session LIMIT ?", (limit,)),
            "sample_committee_members": q(
                "SELECT DISTINCT c.name_raw AS committee, m.full_name AS "
                "member, cm.role, cm.chamber FROM committee_members cm "
                "JOIN committees c ON c.name_key = cm.committee "
                "JOIN members m ON m.name_norm = cm.member "
                "ORDER BY c.name_raw, m.full_name LIMIT ?", (limit,)),
      "committees_without_roster": q(
                  "SELECT name_raw, chamber, session FROM committees c "
                  "WHERE NOT EXISTS (SELECT 1 FROM committee_members cm "
                  "  WHERE cm.committee = c.name_key "
                  "  AND cm.chamber = c.chamber "
                  "  AND cm.session = c.session)"),
        }
    finally:
        conn.close()


# ---------------------------------------------------------------------------
# scheduled_events: the chamber "upcoming events" forward source (v2)
# ---------------------------------------------------------------------------

def _scheduled_events_table(conn: sqlite3.Connection, name: str = "scheduled_events") -> bool:
    cur = conn.execute(
         "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?",
           (name,))
    return cur.fetchone() is not None


def store_scheduled_events(events: Iterable[Dict[str, Any]],
                            path: str = DEFAULT_DB,
                            fetched_at: str = "") -> Dict[str, Any]:
    """Insert parsed chamber events into ``scheduled_events`` idempotently.

    Each ``event`` is a dict with keys ``date`` (ISO), ``chamber`` (upper),
    ``kind`` (``FLOOR``/``COMMITTEE``/``OTHER``), ``title``, ``source_url``,
    ``source``. The natural key ``(date, chamber, kind, title, source)`` makes
    re-runs a no-op (``INSERT OR IGNORE``), so the existing-cron behaviour
    ("re-parse doesn't re-hit") holds: a second parse of the same calendar day
    adds no rows. Ensures the table exists (via :func:`_create_schema`). Returns
    a small summary ``{stored, cleared_stale, total, fetched_at}``.
    """
    conn = connect(path)
    try:
        _create_schema(conn)      # CREATE IF NOT EXISTS; cheap + idempotent
        # Drop a stale day's rows first so a re-published calendar (a moved or
        # renamed hearing) reflects the latest read rather than a frozen
        # snapshot. Only that day is cleared; other days are untouched.
        days = {str(e.get("date")) for e in events if e.get("date")}
        deleted = 0
        for d in days:
            cur = conn.execute(
                "DELETE FROM scheduled_events WHERE date=? AND fetched_at<>?",
                (d, fetched_at))
            deleted += cur.rowcount
        rows = []
        for e in events:
            if not e.get("date") or not e.get("title"):
                continue
            rows.append((
                str(e["date"]).strip(),
                (e.get("chamber") or "unknown").strip().upper(),
                (e.get("kind") or "OTHER").strip().upper() or "OTHER",
                str(e["title"]).strip(),
                e.get("source_url") or None,
                fetched_at or None,
                (e.get("source") or "assembly_events").strip()
                or "assembly_events",
            ))
        if rows:
            cur = conn.executemany(
                "INSERT OR IGNORE INTO scheduled_events "
                "(date, chamber, kind, title, source_url, fetched_at, "
                "source) VALUES (?,?,?,?,?,?,?)",
                rows)
            stored = cur.rowcount
        else:
            stored = 0
        conn.commit()
        total = conn.execute(
            "SELECT COUNT(*) FROM scheduled_events").fetchone()[0]
        return {"stored": stored, "cleared_stale": deleted,
                "total": total, "fetched_at": fetched_at or None}
    finally:
        conn.close()


def load_scheduled_events(path: str = DEFAULT_DB,
                          as_of: Optional[str] = None,
                          chamber: Optional[str] = None,
                          limit: int = 200) -> List[Dict[str, Any]]:
    """Return scheduled (chamber) events, optionally future-dated from ``as_of``.

    A ``None``/absent ``as_of`` keeps every event; a date string keeps
    ``date >= as_of`` (a T-1 event is excluded, matching API_CONTRACT §5).
    Degrades to ``[]`` when the table does not exist (pre-v2 DBs / fresh schema).
    """
    conn = connect(path, foreign_keys=False)
    try:
        if not _scheduled_events_table(conn):
            return []
        sql = ("SELECT date, chamber, kind, title, source_url, fetched_at, "
               "source FROM scheduled_events WHERE 1=1")
        clauses: list = []
        args: list = []
        if as_of:
            clauses.append("date >= ?")
            args.append(str(as_of))
        if chamber:
            clauses.append("UPPER(chamber) = ?")
            args.append(chamber.strip().upper())
        if clauses:
            sql += " AND " + " AND ".join(clauses)
        sql += " ORDER BY date, chamber, kind, title LIMIT ?"
        args.append(int(limit))
        rows = conn.execute(sql, args).fetchall()
    finally:
        conn.close()
    return [dict(r) for r in rows]


def next_legislative_day(path: str = DEFAULT_DB,
                          as_of: Optional[str] = None,
                          chamber: Optional[str] = None) -> Dict[str, Any]:
    """The earliest scheduled event on/after ``as_of`` (DESIGN §6.1 "next
    legislative day"), or a ``{date: None, recess: True}`` marker when nothing is
    scheduled (a genuine recess).

    This is the dashboard's "what's next" answer when the daily-file bill queue
    has no future rows: it surfaces the chamber events calendar. ``recess`` is
    True when there is no event after ``as_of``.
    """
    events = load_scheduled_events(path, as_of=as_of, chamber=chamber, limit=1)
    if not events:
        return {"date": None, "recess": True}
    return {**events[0], "recess": False}