"""Passive same-server inbox metadata on the existing cross-chat SQLite ledger.

Call mutations in the caller's BEGIN IMMEDIATE transaction and current pair
authorization lock. This module never commits, opens a database, starts a turn,
reads credentials, sends a reply, or interprets message text as permission.
"""
from __future__ import annotations

import json
import re
import sqlite3
import uuid
from collections.abc import Iterable


MAX_PAGE_ITEMS = 25
MAX_PAGE_BYTES = 112 * 1024
_PAIR = re.compile(r"pair_[0-9a-f]{32}\Z")


class MailboxConflict(ValueError):
    pass


def _transaction(connection: sqlite3.Connection) -> None:
    if not connection.in_transaction:
        raise RuntimeError("Mailbox mutations require a caller-owned transaction")


def _identifier(value: str) -> str:
    if not isinstance(value, str) or not value or len(value) > 240 or not value.isprintable():
        raise ValueError("Invalid mailbox identifier")
    return value


def _limit(value: int) -> int:
    if type(value) is not int or not 1 <= value <= MAX_PAGE_ITEMS:
        raise ValueError("Mailbox page limit must be between 1 and 25")
    return value


def _cursor(value: int) -> int:
    if type(value) is not int or not 0 <= value <= (1 << 63) - 1:
        raise ValueError("Invalid mailbox cursor")
    return value


def _pairs(values: Iterable[str]) -> str:
    if isinstance(values, (str, bytes)):
        raise ValueError("Expected current authorized pairs")
    clean = set(values)
    if any(not isinstance(value, str) or not _PAIR.fullmatch(value) for value in clean):
        raise ValueError("Invalid authorized pair")
    return json.dumps(sorted(clean), separators=(",", ":"))


def initialize(connection: sqlite3.Connection) -> None:
    """Add mailbox tables without altering existing envelopes or their bodies."""
    statements = (
        """CREATE TABLE IF NOT EXISTS chat_mailbox_reads (
            id TEXT PRIMARY KEY, target_session_id TEXT NOT NULL,
            source_session_id TEXT NOT NULL, reader_run_id TEXT NOT NULL,
            request_id TEXT NOT NULL, snapshot_seq INTEGER NOT NULL,
            page_limit INTEGER NOT NULL, created_at TEXT NOT NULL,
            UNIQUE(target_session_id, reader_run_id, request_id))""",
        """CREATE TABLE IF NOT EXISTS chat_mailbox_messages (
            mailbox_seq INTEGER PRIMARY KEY AUTOINCREMENT,
            message_id TEXT NOT NULL UNIQUE REFERENCES cross_chat_envelopes(id) ON DELETE CASCADE,
            source_session_id TEXT NOT NULL, target_session_id TEXT NOT NULL,
            pair_id TEXT NOT NULL, stored_at TEXT NOT NULL,
            in_reply_to_message_id TEXT, read_id TEXT REFERENCES chat_mailbox_reads(id),
            read_at TEXT, excluded_at TEXT, excluded_reason TEXT,
            read_event_published INTEGER NOT NULL DEFAULT 0 CHECK(read_event_published IN (0,1)))""",
        """CREATE TABLE IF NOT EXISTS chat_mailbox_read_pages (
            read_id TEXT NOT NULL REFERENCES chat_mailbox_reads(id) ON DELETE CASCADE,
            after_seq INTEGER NOT NULL, end_seq INTEGER NOT NULL,
            has_more INTEGER NOT NULL CHECK(has_more IN (0,1)), message_ids_json TEXT NOT NULL,
            PRIMARY KEY(read_id, after_seq))""",
        """CREATE TABLE IF NOT EXISTS chat_mailbox_wakes (
            target_session_id TEXT PRIMARY KEY,
            last_attempt_seq INTEGER NOT NULL DEFAULT 0 CHECK(last_attempt_seq>=0),
            suppressed_seq INTEGER NOT NULL DEFAULT 0 CHECK(suppressed_seq>=0),
            claim_id TEXT UNIQUE, through_seq INTEGER NOT NULL DEFAULT 0 CHECK(through_seq>=0),
            state TEXT CHECK(state IN ('reserved','admitted')), run_id TEXT,
            updated_at TEXT NOT NULL)""",
        """CREATE INDEX IF NOT EXISTS chat_mailbox_sender_order
            ON chat_mailbox_messages(target_session_id, source_session_id, mailbox_seq)""",
        """CREATE INDEX IF NOT EXISTS chat_mailbox_unread_order
            ON chat_mailbox_messages(target_session_id, read_at, mailbox_seq)""",
        """CREATE INDEX IF NOT EXISTS chat_mailbox_read_outbox
            ON chat_mailbox_messages(read_event_published, mailbox_seq)
            WHERE read_at IS NOT NULL AND excluded_at IS NULL""",
        """CREATE INDEX IF NOT EXISTS chat_mailbox_read_request
            ON chat_mailbox_reads(target_session_id, request_id)""",
    )
    for statement in statements:
        connection.execute(statement)


_JOIN = "chat_mailbox_messages m JOIN cross_chat_envelopes e ON e.id=m.message_id"
_VALID = """m.excluded_at IS NULL AND e.status='stored' AND e.delivery_mode='mailbox'
    AND e.kind='instruction' AND e.authorization_kind='configured_route'
    AND e.source_session_id=m.source_session_id AND e.target_session_id=m.target_session_id
    AND e.authorization_pair_id=m.pair_id"""
_COLUMNS = """m.*, e.body AS original_body, e.target_body, e.message_revision, e.created_at,
    e.source_run_id, e.source_user_instruction, e.source_user_delegation_action"""


def _message(row: sqlite3.Row) -> dict:
    edited = row["message_revision"] > 0 and isinstance(row["target_body"], str)
    message = {
        "message_id": row["message_id"], "mailbox_seq": row["mailbox_seq"],
        "source_session_id": row["source_session_id"], "target_session_id": row["target_session_id"],
        "conversation_id": row["pair_id"],
        "body": row["target_body"] if edited else row["original_body"],
        "created_at": row["created_at"], "stored_at": row["stored_at"],
        "in_reply_to_message_id": row["in_reply_to_message_id"],
        "read_id": row["read_id"], "read_at": row["read_at"],
        "message_revision": row["message_revision"] if edited else 0,
        "message_edited_by_user": edited,
    }
    if (row["source_user_delegation_action"] in ("route", "instruction", "request_reply")
            and isinstance(row["source_user_instruction"], str) and row["source_user_instruction"].strip()
            and row["message_revision"] == 0):
        # Only the server's immutable admission marker can attest user origin.
        # An edited body is no longer the originally delegated handoff. Keep
        # it readable, but do not let it inherit the original attestation.
        message["user_delegation"] = {
            "version": 1,
            "source_session_id": row["source_session_id"],
            "source_run_id": row["source_run_id"],
            "target_session_id": row["target_session_id"],
            "reference_action": row["source_user_delegation_action"],
            "source_user_instruction": row["source_user_instruction"],
        }
    else:
        # Retain legacy source text for bounded ledger reads, without attesting
        # it or exposing it as authority in provider/public projections.
        message["source_user_instruction"] = row["source_user_instruction"] or ""
    return message


def store_message(connection: sqlite3.Connection, message_id: str, *, now: str,
                  in_reply_to_message_id: str | None = None) -> dict:
    """Register one already-authorized stored envelope in the same transaction."""
    _transaction(connection)
    record = connection.execute("SELECT * FROM cross_chat_envelopes WHERE id=?",
                                (_identifier(message_id),)).fetchone()
    if (record is None or record["status"] != "stored" or record["delivery_mode"] != "mailbox"
            or record["kind"] != "instruction" or record["authorization_kind"] != "configured_route"
            or not _PAIR.fullmatch(record["authorization_pair_id"] or "")
            or record["source_session_id"] == record["target_session_id"]
            or record["target_run_id"]):
        raise MailboxConflict("Envelope is not an unlaunched mailbox message")
    parent = record["reply_to_message_id"]
    if in_reply_to_message_id is not None and in_reply_to_message_id != parent:
        raise MailboxConflict("Reply relationship does not match the stored envelope")
    if parent:
        prior = connection.execute("SELECT * FROM cross_chat_envelopes WHERE id=?", (parent,)).fetchone()
        prior_mail = connection.execute("SELECT excluded_at FROM chat_mailbox_messages WHERE message_id=?",
                                         (parent,)).fetchone()
        if (prior is None or prior["source_session_id"] != record["target_session_id"]
                or prior["target_session_id"] != record["source_session_id"]
                or prior["authorization_pair_id"] != record["authorization_pair_id"]
                or prior["status"] in ("failed", "cancelled")
                or (prior_mail is not None and prior_mail["excluded_at"] is not None)):
            raise MailboxConflict("Reply must reference an available exact peer message")
    connection.execute("""INSERT OR IGNORE INTO chat_mailbox_messages
        (message_id,source_session_id,target_session_id,pair_id,stored_at,in_reply_to_message_id)
        VALUES(?,?,?,?,?,?)""", (message_id, record["source_session_id"], record["target_session_id"],
                                record["authorization_pair_id"], now, parent))
    row = connection.execute(f"SELECT {_COLUMNS} FROM {_JOIN} WHERE m.message_id=?", (message_id,)).fetchone()
    if (row["source_session_id"] != record["source_session_id"]
            or row["target_session_id"] != record["target_session_id"]
            or row["pair_id"] != record["authorization_pair_id"]
            or row["in_reply_to_message_id"] != parent):
        raise MailboxConflict("Stored mailbox identity changed")
    message = _message(row)
    # Reject before commit if the complete message cannot fit in a bounded read.
    # Preserve all attested source constraints; never truncate or downgrade them.
    projected = {**message, "read_id": "x" * 240, "read_at": "x" * 64}
    try:
        size = len(json.dumps(projected, ensure_ascii=False, separators=(",", ":")).encode("utf-8"))
    except UnicodeEncodeError as exc:
        raise MailboxConflict("Mailbox message must be valid UTF-8") from exc
    if size + 8192 > MAX_PAGE_BYTES:
        raise MailboxConflict("Message and user delegation exceed the bounded mailbox response")
    return message


def _scope(target: str, source: str | None, pairs: Iterable[str]) -> tuple[str, list]:
    clause = f"{_VALID} AND m.target_session_id=? AND m.pair_id IN (SELECT value FROM json_each(?))"
    args = [_identifier(target), _pairs(pairs)]
    if source:
        clause += " AND m.source_session_id=?"
        args.append(_identifier(source))
    return clause, args


def list_senders(connection: sqlite3.Connection, target_session_id: str, allowed_pair_ids: Iterable[str],
                 *, after_sender: str = "", limit: int = MAX_PAGE_ITEMS, unread_only: bool = False) -> dict:
    """Read grouped metadata only; opening the mailbox never marks anything read."""
    clause, args = _scope(target_session_id, None, allowed_pair_ids)
    if after_sender:
        _identifier(after_sender)
    if unread_only:
        clause += " AND m.read_at IS NULL"
    rows = connection.execute(f"""SELECT m.source_session_id, COUNT(*) AS message_count,
        SUM(CASE WHEN m.read_at IS NULL THEN 1 ELSE 0 END) AS unread_count,
        MAX(m.mailbox_seq) AS latest_seq, MAX(e.created_at) AS latest_created_at
        FROM {_JOIN} WHERE {clause} AND m.source_session_id>?
        GROUP BY m.source_session_id ORDER BY m.source_session_id LIMIT ?""",
        (*args, after_sender, _limit(limit) + 1)).fetchall()
    more = len(rows) > limit
    selected = [dict(row) for row in rows[:limit]]
    return {"senders": selected, "has_more": more,
            "next_after_sender": selected[-1]["source_session_id"] if more else None}


def _page(connection: sqlite3.Connection, clause: str, args: list, limit: int,
          read_metadata: dict | None = None) -> tuple[list[dict], bool]:
    messages, used = [], 8192  # reserve for public titles, receipt and reply-route metadata
    rows = connection.execute(f"SELECT {_COLUMNS} FROM {_JOIN} WHERE {clause} ORDER BY m.mailbox_seq LIMIT ?",
                              (*args, limit + 1))
    for row in rows:
        message = _message(row)
        if read_metadata:
            message.update(read_metadata)
        size = len(json.dumps(message, ensure_ascii=False, separators=(",", ":")).encode("utf-8"))
        if len(messages) == limit or used + size > MAX_PAGE_BYTES:
            if not messages:
                raise MailboxConflict("Message exceeds the bounded mailbox response; no message was read")
            return messages, True
        messages.append(message)
        used += size
    return messages, False


def list_messages(connection: sqlite3.Connection, target_session_id: str, source_session_id: str | None,
                  allowed_pair_ids: Iterable[str], *, after_seq: int = 0, limit: int = MAX_PAGE_ITEMS,
                  unread_only: bool = False) -> dict:
    clause, args = _scope(target_session_id, source_session_id, allowed_pair_ids)
    clause += " AND m.mailbox_seq>?"
    args.append(_cursor(after_seq))
    if unread_only:
        clause += " AND m.read_at IS NULL"
    messages, more = _page(connection, clause, args, _limit(limit))
    return {"messages": messages, "has_more": more,
            "next_after_seq": messages[-1]["mailbox_seq"] if more else None}


def read_sender(connection: sqlite3.Connection, *, target_session_id: str, source_session_id: str,
                reader_run_id: str, request_id: str, allowed_pair_ids: Iterable[str], now: str,
                after_seq: int = 0, limit: int = MAX_PAGE_ITEMS) -> dict:
    """Atomically read one page of a fixed unread snapshot, with retry receipts.

    Arrivals after snapshot_seq remain unread. A retry only returns its original
    page IDs across fresh authorized runs, subject to current authorization and
    cancellation/deletion state. The first reader run is attribution, not authority.
    Read means included in this tool response, not proof of model understanding.
    """
    _transaction(connection)
    _identifier(source_session_id)
    _identifier(reader_run_id)
    _identifier(request_id)
    _cursor(after_seq)
    _limit(limit)
    clause, args = _scope(target_session_id, source_session_id, allowed_pair_ids)
    batches = connection.execute("""SELECT * FROM chat_mailbox_reads
        WHERE target_session_id=? AND request_id=? LIMIT 2""",
        (target_session_id, request_id)).fetchall()
    if len(batches) > 1:
        # Older versions permitted this key once per run. Never guess which
        # historical snapshot a caller intended, or consume another batch.
        raise MailboxConflict("Read request matches multiple historical snapshots")
    batch = batches[0] if batches else None
    if batch is None:
        if after_seq:
            raise MailboxConflict("A mailbox read must start at its first page")
        high_water = connection.execute("""SELECT COALESCE(MAX(mailbox_seq),0) FROM chat_mailbox_messages
            WHERE target_session_id=? AND source_session_id=?""", (target_session_id, source_session_id)).fetchone()[0]
        read_id = "mailread_" + uuid.uuid4().hex
        connection.execute("""INSERT INTO chat_mailbox_reads
            (id,target_session_id,source_session_id,reader_run_id,request_id,snapshot_seq,page_limit,created_at)
            VALUES(?,?,?,?,?,?,?,?)""", (read_id, target_session_id, source_session_id, reader_run_id,
                                       request_id, high_water, limit, now))
        batch = connection.execute("SELECT * FROM chat_mailbox_reads WHERE id=?", (read_id,)).fetchone()
    if batch["source_session_id"] != source_session_id or batch["page_limit"] != limit:
        raise MailboxConflict("Read request was already used with different parameters")
    page = connection.execute("SELECT * FROM chat_mailbox_read_pages WHERE read_id=? AND after_seq=?",
                              (batch["id"], after_seq)).fetchone()
    replayed = page is not None
    if page is None:
        if after_seq and connection.execute("""SELECT 1 FROM chat_mailbox_read_pages
            WHERE read_id=? AND end_seq=? AND has_more=1""", (batch["id"], after_seq)).fetchone() is None:
            raise MailboxConflict("Cursor does not belong to this read snapshot")
        messages, more = _page(connection, clause + " AND m.read_at IS NULL AND m.mailbox_seq>? AND m.mailbox_seq<=?",
                               [*args, after_seq, batch["snapshot_seq"]], limit,
                               {"read_id": batch["id"], "read_at": now})
        ids = [message["message_id"] for message in messages]
        end = messages[-1]["mailbox_seq"] if messages else after_seq
        connection.execute("""INSERT INTO chat_mailbox_read_pages
            (read_id,after_seq,end_seq,has_more,message_ids_json) VALUES(?,?,?,?,?)""",
            (batch["id"], after_seq, end, int(more), json.dumps(ids, separators=(",", ":"))))
        for message_id in ids:
            connection.execute("""UPDATE chat_mailbox_messages SET read_id=?,read_at=?
                WHERE message_id=? AND read_at IS NULL""", (batch["id"], now, message_id))
        page = connection.execute("SELECT * FROM chat_mailbox_read_pages WHERE read_id=? AND after_seq=?",
                                  (batch["id"], after_seq)).fetchone()
    ids = json.loads(page["message_ids_json"])
    messages, overflow = _page(connection, clause + " AND m.message_id IN (SELECT value FROM json_each(?))",
                              [*args, json.dumps(ids)], limit) if ids else ([], False)
    if overflow:
        raise MailboxConflict("Stored mailbox read page exceeds the bounded response; retry was not truncated")
    return {"read_id": batch["id"], "snapshot_seq": batch["snapshot_seq"], "messages": messages,
            "next_after_seq": page["end_seq"] if page["has_more"] else None,
            "has_more": bool(page["has_more"]), "unavailable_count": len(ids) - len(messages), "replayed": replayed}


def exclude_message(connection: sqlite3.Connection, message_id: str, *, target_session_id: str,
                    reason: str, now: str) -> bool:
    _transaction(connection)
    if reason not in ("cancelled", "deleted", "revoked"):
        raise ValueError("Invalid mailbox exclusion reason")
    return connection.execute("""UPDATE chat_mailbox_messages SET excluded_at=?,excluded_reason=?
        WHERE message_id=? AND target_session_id=? AND excluded_at IS NULL""",
        (now, reason, _identifier(message_id), _identifier(target_session_id))).rowcount == 1


def cancel_message(connection: sqlite3.Connection, message_id: str, *, now: str) -> bool:
    """Cancel an unread stored message atomically; a completed read wins the race."""
    _transaction(connection)
    record = connection.execute(f"""SELECT m.read_at,m.excluded_at,m.excluded_reason,
        e.status,e.delivery_mode FROM {_JOIN} WHERE m.message_id=?""", (_identifier(message_id),)).fetchone()
    if record is None:
        return False
    if record["status"] == "cancelled" and record["excluded_reason"] == "cancelled":
        return False
    if (record["status"] != "stored" or record["delivery_mode"] != "mailbox"
            or record["excluded_at"] is not None or record["read_at"] is not None):
        raise MailboxConflict("Message is no longer unread and cancellable")
    connection.execute("""UPDATE chat_mailbox_messages SET excluded_at=?,excluded_reason='cancelled'
        WHERE message_id=?""", (now, message_id))
    connection.execute("""UPDATE cross_chat_envelopes SET status='cancelled',lifecycle_status='',updated_at=?
        WHERE id=?""", (now, message_id))
    return True


def exclude_pair(connection: sqlite3.Connection, pair_id: str, *, now: str) -> int:
    _transaction(connection)
    _pairs([pair_id])
    return connection.execute("""UPDATE chat_mailbox_messages SET excluded_at=?,excluded_reason='revoked'
        WHERE pair_id=? AND excluded_at IS NULL""", (now, pair_id)).rowcount


def exclude_session(connection: sqlite3.Connection, session_id: str, *, now: str) -> int:
    _transaction(connection)
    _identifier(session_id)
    return connection.execute("""UPDATE chat_mailbox_messages SET excluded_at=?,excluded_reason='deleted'
        WHERE (source_session_id=? OR target_session_id=?) AND excluded_at IS NULL""",
        (now, session_id, session_id)).rowcount


def pending_read_events(connection: sqlite3.Connection, *, limit: int = MAX_PAGE_ITEMS) -> list[dict]:
    rows, _ = _page(connection, _VALID + " AND m.read_at IS NOT NULL AND m.read_event_published=0", [], _limit(limit))
    return rows


def unread_targets(connection: sqlite3.Connection) -> list[str]:
    """Startup recovery metadata only; never fetch bodies or launch recipients."""
    return [row[0] for row in connection.execute(f"""SELECT DISTINCT m.target_session_id FROM {_JOIN}
        WHERE {_VALID} AND m.read_at IS NULL ORDER BY m.target_session_id""")]


def mark_read_event_published(connection: sqlite3.Connection, message_id: str) -> bool:
    _transaction(connection)
    return connection.execute("""UPDATE chat_mailbox_messages SET read_event_published=1
        WHERE message_id=? AND read_at IS NOT NULL AND read_event_published=0""",
        (_identifier(message_id),)).rowcount == 1


def claim_wake(connection: sqlite3.Connection, target_session_id: str,
               allowed_pair_ids: Iterable[str], *, now: str) -> dict | None:
    """Reserve one authorized unread cutoff; the caller must separately prove idle.

    A wake is an execution attempt, not a read receipt. New arrivals can replace
    an admitted attempt only when the runtime has again proved the chat idle.
    """
    _transaction(connection)
    clause, args = _scope(target_session_id, None, allowed_pair_ids)
    prior = connection.execute("SELECT * FROM chat_mailbox_wakes WHERE target_session_id=?",
                               (target_session_id,)).fetchone()
    if prior is not None and prior["state"] == "reserved":
        return None
    cutoff = max(prior["last_attempt_seq"], prior["suppressed_seq"]) if prior else 0
    through_seq = connection.execute(f"""SELECT COALESCE(MAX(m.mailbox_seq),0) FROM {_JOIN}
        WHERE {clause} AND m.read_at IS NULL AND m.mailbox_seq>?""", (*args, cutoff)).fetchone()[0]
    if not through_seq:
        return None
    claim_id = "mailwake_" + uuid.uuid4().hex
    connection.execute("""INSERT INTO chat_mailbox_wakes
        (target_session_id,claim_id,through_seq,state,updated_at) VALUES(?,?,?,'reserved',?)
        ON CONFLICT(target_session_id) DO UPDATE SET claim_id=excluded.claim_id,
        through_seq=excluded.through_seq,state='reserved',run_id=NULL,updated_at=excluded.updated_at""",
        (target_session_id, claim_id, through_seq, now))
    return dict(connection.execute("SELECT * FROM chat_mailbox_wakes WHERE target_session_id=?",
                                   (target_session_id,)).fetchone())


def admit_wake(connection: sqlite3.Connection, target_session_id: str, claim_id: str,
               run_id: str, allowed_pair_ids: Iterable[str], *, now: str) -> bool:
    """Consume one reservation before provider launch, retaining ambiguous attempts.

    Current route permission and unread state are rechecked within the claimed
    cutoff. Even an identical repeated admission returns False: it is not a
    second authorization to launch the provider.
    """
    _transaction(connection)
    clause, args = _scope(target_session_id, None, allowed_pair_ids)
    _identifier(claim_id)
    _identifier(run_id)
    claim = connection.execute("""SELECT * FROM chat_mailbox_wakes
        WHERE target_session_id=? AND claim_id=? AND state='reserved'""",
        (target_session_id, claim_id)).fetchone()
    if claim is None:
        return False
    eligible = connection.execute(f"""SELECT 1 FROM {_JOIN} WHERE {clause}
        AND m.read_at IS NULL AND m.mailbox_seq>? AND m.mailbox_seq<=? LIMIT 1""",
        (*args, max(claim["last_attempt_seq"], claim["suppressed_seq"]), claim["through_seq"])).fetchone()
    if eligible is None:
        return False
    return connection.execute("""UPDATE chat_mailbox_wakes SET state='admitted',run_id=?,
        last_attempt_seq=MAX(last_attempt_seq,through_seq),updated_at=?
        WHERE target_session_id=? AND claim_id=? AND state='reserved'""",
        (run_id, now, target_session_id, claim_id)).rowcount == 1


def release_wake(connection: sqlite3.Connection, target_session_id: str, claim_id: str) -> bool:
    """Release only a known unlaunched reservation, never an admitted attempt."""
    _transaction(connection)
    return connection.execute("""UPDATE chat_mailbox_wakes
        SET claim_id=NULL,through_seq=0,state=NULL,run_id=NULL
        WHERE target_session_id=? AND claim_id=? AND state='reserved'""",
        (_identifier(target_session_id), _identifier(claim_id))).rowcount == 1


def suppress_wake(connection: sqlite3.Connection, target_session_id: str, *, now: str) -> int:
    """Stop suppresses current unread mail, without reading it or future arrivals."""
    _transaction(connection)
    target_session_id = _identifier(target_session_id)
    through_seq = connection.execute(f"""SELECT COALESCE(MAX(m.mailbox_seq),0) FROM {_JOIN}
        WHERE {_VALID} AND m.target_session_id=? AND m.read_at IS NULL""",
        (target_session_id,)).fetchone()[0]
    connection.execute("""INSERT INTO chat_mailbox_wakes
        (target_session_id,suppressed_seq,updated_at) VALUES(?,?,?)
        ON CONFLICT(target_session_id) DO UPDATE
        SET suppressed_seq=MAX(suppressed_seq,excluded.suppressed_seq),updated_at=excluded.updated_at""",
        (target_session_id, through_seq, now))
    connection.execute("""UPDATE chat_mailbox_wakes
        SET claim_id=NULL,through_seq=0,state=NULL,run_id=NULL
        WHERE target_session_id=? AND state='reserved'""", (target_session_id,))
    return connection.execute("SELECT suppressed_seq FROM chat_mailbox_wakes WHERE target_session_id=?",
                              (target_session_id,)).fetchone()[0]


def recover_wakes(connection: sqlite3.Connection) -> int:
    """Startup only: no provider may launch before its reservation is admitted."""
    _transaction(connection)
    return connection.execute("""UPDATE chat_mailbox_wakes
        SET claim_id=NULL,through_seq=0,state=NULL,run_id=NULL WHERE state='reserved'""").rowcount