"""Durable, account-scoped ordered Today plans."""

import json
import sqlite3
import time
from datetime import date, datetime
from pathlib import Path
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

from src.private_state import connect_private_sqlite
from src.state_encryption import PrivateStateCipher, PrivateStateEncryptionError, private_state_encryption_key


class TodayPlanFull(ValueError):
    """Raised when an add would exceed the bounded Today plan."""


class TodaySessionConflict(ValueError):
    """Raised when a device updates an obsolete active-session revision."""

    def __init__(self, session: dict):
        super().__init__("active Today session changed on another device")
        self.session = session


class TomorrowPlanConflict(ValueError):
    """Raised when Tomorrow was edited from an obsolete revision."""

    def __init__(self, snapshot: dict):
        super().__init__("Tomorrow plan changed on another device")
        self.snapshot = snapshot


class TomorrowPlanNotDue(ValueError):
    """Raised when Tomorrow is promoted before its saved local date."""

    def __init__(self, plan_date: str):
        super().__init__("Tomorrow plan is not due")
        self.plan_date = plan_date


class WeekPlanConflict(ValueError):
    """Raised when Week Ahead was edited from an obsolete revision."""

    def __init__(self, snapshot: dict):
        super().__init__("Week Ahead changed on another device")
        self.snapshot = snapshot


class TodayPromotionConflict(ValueError):
    """Raised when promotion cannot safely replace the current Today plan."""

    def __init__(self, today: dict, week: dict | None = None):
        super().__init__("Today must be reviewed before a saved plan can be promoted")
        self.today = today
        self.week = week


class TodayStore:
    def __init__(
        self,
        path: str | Path,
        *,
        limit: int = 5,
        timeout: float = 1.0,
        operation_limit: int = 4096,
        operation_retention_seconds: float = 30 * 24 * 60 * 60,
        recap_limit: int = 100,
        encryption_key: bytes | None = None,
        clock=time.time,
    ):
        self.path = Path(path)
        self.limit = limit
        self.timeout = timeout
        self.operation_limit = operation_limit
        self.operation_retention_seconds = operation_retention_seconds
        self.recap_limit = recap_limit
        self.clock = clock
        self._cipher = PrivateStateCipher(
            encryption_key if encryption_key is not None else private_state_encryption_key(),
            store="today",
        )
        self._initialize()

    def _initialize(self) -> None:
        connection = connect_private_sqlite(self.path, timeout=self.timeout)
        if connection.execute("PRAGMA user_version").fetchone()[0] >= 5:
            connection.close()
            return
        connection.execute("PRAGMA journal_mode=WAL")
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS today_plans (
                login TEXT PRIMARY KEY,
                revision INTEGER NOT NULL,
                ids TEXT NOT NULL,
                capacity_minutes INTEGER,
                estimates TEXT NOT NULL DEFAULT '{}',
                plan_date TEXT,
                timezone TEXT
            )
            """
        )
        plan_columns = {row[1] for row in connection.execute("PRAGMA table_info(today_plans)")}
        if "capacity_minutes" not in plan_columns:
            connection.execute("ALTER TABLE today_plans ADD COLUMN capacity_minutes INTEGER")
        if "estimates" not in plan_columns:
            connection.execute("ALTER TABLE today_plans ADD COLUMN estimates TEXT NOT NULL DEFAULT '{}'")
        if "plan_date" not in plan_columns:
            connection.execute("ALTER TABLE today_plans ADD COLUMN plan_date TEXT")
        if "timezone" not in plan_columns:
            connection.execute("ALTER TABLE today_plans ADD COLUMN timezone TEXT")
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS today_operations (
                login TEXT NOT NULL,
                operation_id TEXT NOT NULL,
                created_at REAL NOT NULL,
                PRIMARY KEY (login, operation_id)
            )
            """
        )
        columns = {row[1] for row in connection.execute("PRAGMA table_info(today_operations)")}
        if "created_at" not in columns:
            connection.execute("ALTER TABLE today_operations ADD COLUMN created_at REAL")
        connection.execute(
            "UPDATE today_operations SET created_at = ? WHERE created_at IS NULL",
            (self.clock(),),
        )
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS today_recaps (
                login TEXT NOT NULL,
                session_id TEXT NOT NULL,
                created_at REAL NOT NULL,
                items TEXT NOT NULL,
                PRIMARY KEY (login, session_id)
            )
            """
        )
        connection.execute(
            "CREATE INDEX IF NOT EXISTS today_recaps_recent "
            "ON today_recaps(login, created_at DESC)"
        )
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS today_time_logs (
                login TEXT NOT NULL,
                session_id TEXT NOT NULL,
                identity TEXT NOT NULL,
                actual_minutes INTEGER NOT NULL,
                status TEXT NOT NULL,
                PRIMARY KEY (login, session_id, identity)
            )
            """
        )
        connection.execute(
            """
            CREATE TABLE IF NOT EXISTS today_sessions (
                login TEXT PRIMARY KEY,
                revision INTEGER NOT NULL,
                device_id TEXT NOT NULL,
                identity TEXT NOT NULL,
                elapsed_ms INTEGER NOT NULL,
                running INTEGER NOT NULL,
                break_deadline_at INTEGER,
                updated_at REAL NOT NULL
            )
            """
        )
        session_columns = {row[1] for row in connection.execute("PRAGMA table_info(today_sessions)")}
        if "break_deadline_at" not in session_columns:
            connection.execute("ALTER TABLE today_sessions ADD COLUMN break_deadline_at INTEGER")
        connection.execute(
            "CREATE TABLE IF NOT EXISTS tomorrow_plans ("
            "login TEXT PRIMARY KEY, revision INTEGER NOT NULL, payload TEXT NOT NULL)"
        )
        connection.execute(
            "CREATE TABLE IF NOT EXISTS tomorrow_promotions ("
            "login TEXT NOT NULL, promotion_id TEXT NOT NULL, result TEXT NOT NULL, "
            "created_at REAL NOT NULL, PRIMARY KEY (login, promotion_id))"
        )
        connection.execute(
            "CREATE TABLE IF NOT EXISTS week_plans ("
            "login TEXT PRIMARY KEY, revision INTEGER NOT NULL, payload TEXT NOT NULL)"
        )
        connection.execute(
            "CREATE TABLE IF NOT EXISTS week_promotions ("
            "login TEXT NOT NULL, promotion_id TEXT NOT NULL, result TEXT NOT NULL, "
            "created_at REAL NOT NULL, PRIMARY KEY (login, promotion_id))"
        )
        connection.execute("PRAGMA user_version = 5")
        connection.commit()
        connection.close()

    def _connect(self) -> sqlite3.Connection:
        return connect_private_sqlite(self.path, timeout=self.timeout)

    def _record_operation(self, connection: sqlite3.Connection, login: str, operation_id: str) -> None:
        now = self.clock()
        connection.execute(
            "INSERT INTO today_operations(login, operation_id, created_at) VALUES (?, ?, ?)",
            (login, operation_id, now),
        )
        connection.execute(
            "DELETE FROM today_operations WHERE login = ? AND created_at < ?",
            (login, now - self.operation_retention_seconds),
        )
        connection.execute(
            "DELETE FROM today_operations WHERE login = ? AND rowid NOT IN "
            "(SELECT rowid FROM today_operations WHERE login = ? "
            "ORDER BY created_at DESC, rowid DESC LIMIT ?)",
            (login, login, self.operation_limit),
        )

    @staticmethod
    def _normalize_login(login: str) -> str:
        normalized = login.strip().lower()
        if not normalized:
            raise ValueError("login is required")
        return normalized

    def _snapshot(self, row, login: str) -> tuple[dict, bool]:
        if row is None:
            return {"revision": 0, "ids": [], "capacity_minutes": None, "estimates": {}}, False
        if row[1].startswith("v1:"):
            payload, legacy = self._cipher.open(row[1], binding=f"plan:{login}")
            if not isinstance(payload, dict):
                raise PrivateStateEncryptionError("private state could not be decrypted")
            ids = payload.get("ids")
            estimates = payload.get("estimates")
            capacity_minutes = payload.get("capacity_minutes")
            plan_date = payload.get("plan_date")
            timezone = payload.get("timezone")
            first_task_state = payload.get("first_task_state", "")
        else:
            legacy = True
            ids = json.loads(row[1])
            estimates = json.loads(row[3] or "{}")
            capacity_minutes = row[2]
            plan_date = row[4] if len(row) > 4 else None
            timezone = row[5] if len(row) > 5 else None
            first_task_state = ""
        if not isinstance(ids, list) or not isinstance(estimates, dict):
            raise PrivateStateEncryptionError("private state could not be decrypted")
        snapshot = {
            "revision": int(row[0]),
            "ids": ids,
            "capacity_minutes": capacity_minutes,
            "estimates": {item_id: minutes for item_id, minutes in estimates.items() if item_id in ids},
        }
        if plan_date:
            snapshot["plan_date"] = plan_date
            snapshot["timezone"] = timezone
        if first_task_state in {"coaching", "complete"}:
            snapshot["first_task_state"] = first_task_state
        return snapshot, legacy

    def _sealed_plan(self, login: str, snapshot: dict) -> str:
        return self._cipher.seal({
            "ids": snapshot["ids"],
            "capacity_minutes": snapshot["capacity_minutes"],
            "estimates": snapshot["estimates"],
            "plan_date": snapshot.get("plan_date"),
            "timezone": snapshot.get("timezone"),
            "first_task_state": snapshot.get("first_task_state", ""),
        }, binding=f"plan:{login}")

    def get(self, login: str) -> dict:
        login = self._normalize_login(login)
        with self._connect() as connection:
            row = connection.execute(
                "SELECT revision, ids, capacity_minutes, estimates, plan_date, timezone "
                "FROM today_plans WHERE login = ?",
                (login,),
            ).fetchone()
            snapshot, legacy = self._snapshot(row, login)
            if row is not None and legacy:
                connection.execute(
                    "UPDATE today_plans SET ids = ?, capacity_minutes = NULL, estimates = '{}', "
                    "plan_date = NULL, timezone = NULL WHERE login = ? AND ids = ?",
                    (self._sealed_plan(login, snapshot), login, row[1]),
                )
        return snapshot

    @staticmethod
    def _empty_tomorrow(revision: int = 0) -> dict:
        return {"revision": revision, "ids": [], "capacity_minutes": None, "estimates": {}}

    def _tomorrow_snapshot(self, row, login: str) -> dict:
        if row is None:
            return self._empty_tomorrow()
        payload, _legacy = self._cipher.open(row[1], binding=f"tomorrow:{login}")
        if not isinstance(payload, dict):
            raise PrivateStateEncryptionError("private state could not be decrypted")
        snapshot = {"revision": int(row[0]), **payload}
        if not isinstance(snapshot.get("ids"), list) or not isinstance(snapshot.get("estimates"), dict):
            raise PrivateStateEncryptionError("private state could not be decrypted")
        return snapshot

    def get_tomorrow(self, login: str) -> dict:
        login = self._normalize_login(login)
        with self._connect() as connection:
            row = connection.execute(
                "SELECT revision, payload FROM tomorrow_plans WHERE login = ?", (login,)
            ).fetchone()
            return self._tomorrow_snapshot(row, login)

    def _normalize_tomorrow(
        self, *, ids: list[str], capacity_minutes: int | None,
        estimates: dict[str, int], plan_date: str, timezone: str,
    ) -> dict:
        try:
            if date.fromisoformat(plan_date or "").isoformat() != plan_date:
                raise ValueError
        except (TypeError, ValueError):
            raise ValueError("plan_date must be an ISO calendar date") from None
        if not isinstance(timezone, str) or not timezone.strip() or len(timezone) > 100:
            raise ValueError("timezone is required and bounded")
        timezone = timezone.strip()
        try:
            ZoneInfo(timezone)
        except (ZoneInfoNotFoundError, ValueError):
            raise ValueError("timezone must be a valid IANA timezone") from None
        if not isinstance(ids, list) or len(ids) > self.limit or any(
            not isinstance(item_id, str) or not item_id or len(item_id) > 500 for item_id in ids
        ):
            raise ValueError("Tomorrow IDs are invalid or exceed the plan limit")
        if len(set(ids)) != len(ids):
            raise ValueError("Tomorrow IDs must be unique")
        if capacity_minutes is not None and (
            not isinstance(capacity_minutes, int) or isinstance(capacity_minutes, bool)
            or capacity_minutes < 15 or capacity_minutes > 1440
        ):
            raise ValueError("capacity_minutes must be between 15 and 1440")
        if not isinstance(estimates, dict):
            raise ValueError("estimates must be an object")
        normalized_estimates = {}
        for item_id, minutes in estimates.items():
            if item_id not in ids:
                continue
            if not isinstance(minutes, int) or isinstance(minutes, bool) or minutes < 5 or minutes > 1440:
                raise ValueError("estimate minutes must be between 5 and 1440")
            normalized_estimates[item_id] = minutes
        return {
            "ids": list(ids), "capacity_minutes": capacity_minutes,
            "estimates": normalized_estimates, "plan_date": plan_date,
            "timezone": timezone.strip(),
        }

    def replace_tomorrow(self, login: str, *, base_revision: int, **plan) -> dict:
        login = self._normalize_login(login)
        if not isinstance(base_revision, int) or isinstance(base_revision, bool) or base_revision < 0:
            raise ValueError("base_revision must be a non-negative integer")
        normalized = self._normalize_tomorrow(**plan)
        with self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            row = connection.execute(
                "SELECT revision, payload FROM tomorrow_plans WHERE login = ?", (login,)
            ).fetchone()
            current = self._tomorrow_snapshot(row, login)
            if current["revision"] != base_revision:
                raise TomorrowPlanConflict(current)
            snapshot = {"revision": base_revision + 1, **normalized}
            sealed = self._cipher.seal(normalized, binding=f"tomorrow:{login}")
            connection.execute(
                "INSERT INTO tomorrow_plans(login, revision, payload) VALUES (?, ?, ?) "
                "ON CONFLICT(login) DO UPDATE SET revision=excluded.revision, payload=excluded.payload",
                (login, snapshot["revision"], sealed),
            )
            return snapshot

    def promote_tomorrow(
        self, login: str, *, promotion_id: str, tomorrow_revision: int, today_revision: int,
    ) -> dict:
        login = self._normalize_login(login)
        if not isinstance(promotion_id, str) or not promotion_id.strip() or len(promotion_id) > 100:
            raise ValueError("promotion_id is required and bounded")
        promotion_id = promotion_id.strip()
        with self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            receipt = connection.execute(
                "SELECT result FROM tomorrow_promotions WHERE login = ? AND promotion_id = ?",
                (login, promotion_id),
            ).fetchone()
            if receipt:
                result, _legacy = self._cipher.open(
                    receipt[0], binding=f"tomorrow-promotion:{login}:{promotion_id}"
                )
                if not isinstance(result, dict):
                    raise PrivateStateEncryptionError("private state could not be decrypted")
                return result
            tomorrow_row = connection.execute(
                "SELECT revision, payload FROM tomorrow_plans WHERE login = ?", (login,)
            ).fetchone()
            tomorrow = self._tomorrow_snapshot(tomorrow_row, login)
            if tomorrow["revision"] != tomorrow_revision or not tomorrow.get("plan_date"):
                raise TomorrowPlanConflict(tomorrow)
            local_date = datetime.fromtimestamp(
                self.clock(), ZoneInfo(tomorrow["timezone"])
            ).date().isoformat()
            if local_date < tomorrow["plan_date"]:
                raise TomorrowPlanNotDue(tomorrow["plan_date"])
            today_row = connection.execute(
                "SELECT revision, ids, capacity_minutes, estimates, plan_date, timezone "
                "FROM today_plans WHERE login = ?", (login,)
            ).fetchone()
            today, _legacy = self._snapshot(today_row, login)
            if today["revision"] != today_revision:
                raise TodayPromotionConflict(today)
            result = {"revision": today_revision + 1, **{
                key: tomorrow[key] for key in
                ("ids", "capacity_minutes", "estimates", "plan_date", "timezone")
            }}
            sealed_today = self._sealed_plan(login, result)
            connection.execute(
                "INSERT INTO today_plans(login, revision, ids, capacity_minutes, estimates, plan_date, timezone) "
                "VALUES (?, ?, ?, NULL, '{}', NULL, NULL) ON CONFLICT(login) DO UPDATE SET "
                "revision=excluded.revision, ids=excluded.ids, capacity_minutes=NULL, estimates='{}', "
                "plan_date=NULL, timezone=NULL",
                (login, result["revision"], sealed_today),
            )
            empty = self._empty_tomorrow(tomorrow_revision + 1)
            sealed_empty = self._cipher.seal(
                {key: empty[key] for key in ("ids", "capacity_minutes", "estimates")},
                binding=f"tomorrow:{login}",
            )
            connection.execute(
                "UPDATE tomorrow_plans SET revision = ?, payload = ? WHERE login = ?",
                (empty["revision"], sealed_empty, login),
            )
            sealed_result = self._cipher.seal(
                result, binding=f"tomorrow-promotion:{login}:{promotion_id}"
            )
            connection.execute(
                "INSERT INTO tomorrow_promotions(login, promotion_id, result, created_at) VALUES (?, ?, ?, ?)",
                (login, promotion_id, sealed_result, self.clock()),
            )
            return result

    @staticmethod
    def _empty_week(revision: int = 0) -> dict:
        return {"revision": revision, "timezone": None, "days": []}

    def _week_snapshot(self, row, login: str) -> dict:
        if row is None:
            return self._empty_week()
        payload, _legacy = self._cipher.open(row[1], binding=f"week:{login}")
        if not isinstance(payload, dict) or not isinstance(payload.get("days"), list):
            raise PrivateStateEncryptionError("private state could not be decrypted")
        return {"revision": int(row[0]), **payload}

    def get_week(self, login: str) -> dict:
        login = self._normalize_login(login)
        with self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            row = connection.execute(
                "SELECT revision, payload FROM week_plans WHERE login = ?", (login,)
            ).fetchone()
            if row is not None:
                return self._week_snapshot(row, login)
            tomorrow_row = connection.execute(
                "SELECT revision, payload FROM tomorrow_plans WHERE login = ?", (login,)
            ).fetchone()
            tomorrow = self._tomorrow_snapshot(tomorrow_row, login)
            if not tomorrow.get("plan_date"):
                return self._empty_week()
            day = {key: tomorrow[key] for key in (
                "plan_date", "ids", "capacity_minutes", "estimates"
            )}
            payload = {"timezone": tomorrow["timezone"], "days": [day]}
            connection.execute(
                "INSERT INTO week_plans(login, revision, payload) VALUES (?, 1, ?)",
                (login, self._cipher.seal(payload, binding=f"week:{login}")),
            )
            empty = self._empty_tomorrow(tomorrow["revision"] + 1)
            connection.execute(
                "UPDATE tomorrow_plans SET revision = ?, payload = ? WHERE login = ?",
                (empty["revision"], self._cipher.seal(
                    {key: empty[key] for key in ("ids", "capacity_minutes", "estimates")},
                    binding=f"tomorrow:{login}",
                ), login),
            )
            return {"revision": 1, **payload}

    def _normalize_week(self, *, days: list[dict], timezone: str) -> dict:
        if not isinstance(days, list) or len(days) > 7:
            raise ValueError("Week Ahead is limited to seven dates")
        if not isinstance(timezone, str) or not timezone.strip() or len(timezone) > 100:
            raise ValueError("timezone is required and bounded")
        timezone = timezone.strip()
        try:
            ZoneInfo(timezone)
        except (ZoneInfoNotFoundError, ValueError):
            raise ValueError("timezone must be a valid IANA timezone") from None
        normalized = []
        seen_dates = set()
        for day in days:
            if not isinstance(day, dict):
                raise ValueError("each Week Ahead date must be an object")
            plan_date = day.get("plan_date")
            try:
                if date.fromisoformat(plan_date or "").isoformat() != plan_date:
                    raise ValueError
            except (TypeError, ValueError):
                raise ValueError("plan_date must be an ISO calendar date") from None
            if plan_date in seen_dates:
                raise ValueError("Week Ahead dates must be unique")
            seen_dates.add(plan_date)
            plan = self._normalize_tomorrow(
                ids=day.get("ids", []), capacity_minutes=day.get("capacity_minutes"),
                estimates=day.get("estimates", {}), plan_date=plan_date, timezone=timezone,
            )
            plan.pop("timezone")
            normalized.append(plan)
        normalized.sort(key=lambda item: item["plan_date"])
        return {"timezone": timezone, "days": normalized}

    def replace_week(self, login: str, *, base_revision: int, days: list[dict], timezone: str) -> dict:
        login = self._normalize_login(login)
        if not isinstance(base_revision, int) or isinstance(base_revision, bool) or base_revision < 0:
            raise ValueError("base_revision must be a non-negative integer")
        normalized = self._normalize_week(days=days, timezone=timezone)
        with self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            row = connection.execute(
                "SELECT revision, payload FROM week_plans WHERE login = ?", (login,)
            ).fetchone()
            current = self._week_snapshot(row, login)
            if current["revision"] != base_revision:
                raise WeekPlanConflict(current)
            snapshot = {"revision": base_revision + 1, **normalized}
            connection.execute(
                "INSERT INTO week_plans(login, revision, payload) VALUES (?, ?, ?) "
                "ON CONFLICT(login) DO UPDATE SET revision=excluded.revision, payload=excluded.payload",
                (login, snapshot["revision"], self._cipher.seal(normalized, binding=f"week:{login}")),
            )
            return snapshot

    def promote_week(
        self, login: str, *, promotion_id: str, week_revision: int,
        plan_date: str, today_revision: int,
    ) -> dict:
        login = self._normalize_login(login)
        if not isinstance(promotion_id, str) or not promotion_id.strip() or len(promotion_id) > 100:
            raise ValueError("promotion_id is required and bounded")
        promotion_id = promotion_id.strip()
        with self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            receipt = connection.execute(
                "SELECT result FROM week_promotions WHERE login = ? AND promotion_id = ?",
                (login, promotion_id),
            ).fetchone()
            if receipt:
                result, _legacy = self._cipher.open(
                    receipt[0], binding=f"week-promotion:{login}:{promotion_id}"
                )
                if not isinstance(result, dict):
                    raise PrivateStateEncryptionError("private state could not be decrypted")
                return result
            row = connection.execute(
                "SELECT revision, payload FROM week_plans WHERE login = ?", (login,)
            ).fetchone()
            week = self._week_snapshot(row, login)
            if week["revision"] != week_revision:
                raise WeekPlanConflict(week)
            day = next((item for item in week["days"] if item["plan_date"] == plan_date), None)
            if day is None:
                raise WeekPlanConflict(week)
            local_date = datetime.fromtimestamp(self.clock(), ZoneInfo(week["timezone"])).date().isoformat()
            if local_date < plan_date:
                raise TomorrowPlanNotDue(plan_date)
            today_row = connection.execute(
                "SELECT revision, ids, capacity_minutes, estimates, plan_date, timezone "
                "FROM today_plans WHERE login = ?", (login,)
            ).fetchone()
            today, _legacy = self._snapshot(today_row, login)
            if today["revision"] != today_revision:
                raise TodayPromotionConflict(today)
            if today["ids"]:
                raise TodayPromotionConflict(today, week)
            result = {"revision": today_revision + 1, **day, "timezone": week["timezone"]}
            connection.execute(
                "INSERT INTO today_plans(login, revision, ids, capacity_minutes, estimates, plan_date, timezone) "
                "VALUES (?, ?, ?, NULL, '{}', NULL, NULL) ON CONFLICT(login) DO UPDATE SET "
                "revision=excluded.revision, ids=excluded.ids, capacity_minutes=NULL, estimates='{}', "
                "plan_date=NULL, timezone=NULL",
                (login, result["revision"], self._sealed_plan(login, result)),
            )
            remaining = {"timezone": week["timezone"], "days": [
                item for item in week["days"] if item["plan_date"] != plan_date
            ]}
            connection.execute(
                "UPDATE week_plans SET revision = ?, payload = ? WHERE login = ?",
                (week_revision + 1, self._cipher.seal(remaining, binding=f"week:{login}"), login),
            )
            connection.execute(
                "INSERT INTO week_promotions(login, promotion_id, result, created_at) VALUES (?, ?, ?, ?)",
                (login, promotion_id, self._cipher.seal(
                    result, binding=f"week-promotion:{login}:{promotion_id}"
                ), self.clock()),
            )
            return result

    @staticmethod
    def _empty_session() -> dict:
        return {
            "revision": 0, "device_id": "", "identity": "",
            "elapsed_ms": 0, "running": False, "break_deadline_at": None, "updated_at": None,
        }

    def _session_snapshot(self, row, login: str) -> tuple[dict, bool]:
        if row is None:
            return self._empty_session(), False
        if row[1].startswith("v1:"):
            payload, legacy = self._cipher.open(row[1], binding=f"session:{login}")
            if not isinstance(payload, dict):
                raise PrivateStateEncryptionError("private state could not be decrypted")
            return {"revision": int(row[0]), **payload, "updated_at": row[6]}, legacy
        return {
            "revision": int(row[0]), "device_id": row[1], "identity": row[2],
            "elapsed_ms": int(row[3]), "running": bool(row[4]),
            "break_deadline_at": row[5], "updated_at": row[6],
        }, True

    def _sealed_session(self, login: str, session: dict) -> str:
        return self._cipher.seal({
            "device_id": session["device_id"], "identity": session["identity"],
            "elapsed_ms": session["elapsed_ms"], "running": session["running"],
            "break_deadline_at": session["break_deadline_at"],
        }, binding=f"session:{login}")

    def get_session(self, login: str) -> dict:
        login = self._normalize_login(login)
        with self._connect() as connection:
            row = connection.execute(
                "SELECT revision, device_id, identity, elapsed_ms, running, break_deadline_at, updated_at "
                "FROM today_sessions WHERE login = ?", (login,),
            ).fetchone()
            session, legacy = self._session_snapshot(row, login)
            if row is not None and legacy:
                connection.execute(
                    "UPDATE today_sessions SET device_id = ?, identity = '', elapsed_ms = 0, "
                    "running = 0, break_deadline_at = NULL WHERE login = ? AND device_id = ?",
                    (self._sealed_session(login, session), login, row[1]),
                )
        return session

    def update_session(
        self, login: str, *, base_revision: int, device_id: str,
        identity: str, elapsed_ms: int, running: bool, break_deadline_at: int | None = None,
    ) -> dict:
        login = self._normalize_login(login)
        updated_at = self.clock()
        with self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            current = connection.execute(
                "SELECT revision, device_id, identity, elapsed_ms, running, break_deadline_at, updated_at "
                "FROM today_sessions WHERE login = ?", (login,)
            ).fetchone()
            current_session, _legacy = self._session_snapshot(current, login)
            current_revision = current_session["revision"]
            if base_revision != current_revision:
                raise TodaySessionConflict(current_session)
            revision = current_revision + 1
            session = {
                "revision": revision, "device_id": device_id, "identity": identity,
                "elapsed_ms": elapsed_ms, "running": bool(running),
                "break_deadline_at": break_deadline_at, "updated_at": updated_at,
            }
            connection.execute(
                "INSERT INTO today_sessions(login, revision, device_id, identity, elapsed_ms, running, break_deadline_at, updated_at) "
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?) "
                "ON CONFLICT(login) DO UPDATE SET revision=excluded.revision, device_id=excluded.device_id, "
                "identity=excluded.identity, elapsed_ms=excluded.elapsed_ms, running=excluded.running, "
                "break_deadline_at=excluded.break_deadline_at, updated_at=excluded.updated_at",
                (login, revision, self._sealed_session(login, session), "", 0, 0, None, updated_at),
            )
        return session

    @staticmethod
    def _normalize_recap_items(items: list[dict]) -> list[dict]:
        if not isinstance(items, list) or not items:
            raise ValueError("recap requires at least one item")
        if len(items) > 20:
            raise ValueError("recap is limited to 20 items")
        normalized = []
        seen = set()
        for item in items:
            identity = item.get("identity", "") if isinstance(item, dict) else ""
            if not isinstance(identity, str) or not identity.strip() or len(identity) > 500:
                raise ValueError("recap item identity is required and bounded")
            identity = identity.strip()
            if identity in seen:
                raise ValueError("recap item identities must be unique")
            actual = item.get("actual_minutes")
            estimate = item.get("estimate_minutes")
            if not isinstance(actual, int) or isinstance(actual, bool) or actual < 0 or actual > 1440:
                raise ValueError("actual minutes must be between 0 and 1440")
            if estimate is not None and (
                not isinstance(estimate, int) or isinstance(estimate, bool)
                or estimate < 5 or estimate > 1440
            ):
                raise ValueError("estimate minutes must be between 5 and 1440")
            seen.add(identity)
            normalized.append({
                "identity": identity,
                "estimate_minutes": estimate,
                "actual_minutes": actual,
            })
        return normalized

    def _recap_snapshot(
        self, login: str, encrypted_session_id: str, created_at: float, serialized: str
    ) -> tuple[dict, bool]:
        if encrypted_session_id.startswith("v1:"):
            session_id, legacy_session = self._cipher.open(
                encrypted_session_id, binding=f"recap-id:{login}"
            )
            items, legacy_items = self._cipher.open(
                serialized, binding=f"recap-items:{login}:{session_id}"
            )
        else:
            session_id, legacy_session = encrypted_session_id, True
            try:
                items, legacy_items = json.loads(serialized), True
            except json.JSONDecodeError as error:
                raise PrivateStateEncryptionError("private state could not be decrypted") from error
        if not isinstance(session_id, str) or not isinstance(items, list):
            raise PrivateStateEncryptionError("private state could not be decrypted")
        estimated = sum(item["estimate_minutes"] for item in items if item["estimate_minutes"] is not None)
        actual = sum(item["actual_minutes"] for item in items)
        return {
            "session_id": session_id,
            "created_at": created_at,
            "items": items,
            "estimated_minutes": estimated,
            "actual_minutes": actual,
            "variance_minutes": actual - estimated,
        }, legacy_session or legacy_items

    def _migrate_recap(self, connection, login: str, stored_session: str, snapshot: dict) -> None:
        encrypted_session = self._cipher.seal(
            snapshot["session_id"], binding=f"recap-id:{login}"
        )
        encrypted_items = self._cipher.seal(
            snapshot["items"], binding=f"recap-items:{login}:{snapshot['session_id']}"
        )
        connection.execute(
            "UPDATE today_recaps SET session_id = ?, items = ? "
            "WHERE login = ? AND session_id = ?",
            (encrypted_session, encrypted_items, login, stored_session),
        )

    def save_recap(self, login: str, session_id: str, items: list[dict]) -> dict:
        login = self._normalize_login(login)
        if not isinstance(session_id, str) or not session_id.strip() or len(session_id) > 100:
            raise ValueError("session_id is required and bounded")
        session_id = session_id.strip()
        normalized = self._normalize_recap_items(items)
        with self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            existing_rows = connection.execute(
                "SELECT session_id, created_at, items FROM today_recaps WHERE login = ?", (login,)
            ).fetchall()
            for existing in existing_rows:
                snapshot, legacy = self._recap_snapshot(login, *existing)
                if snapshot["session_id"] == session_id:
                    if legacy:
                        self._migrate_recap(connection, login, existing[0], snapshot)
                    return snapshot
            created_at = self.clock()
            encrypted_session_id = self._cipher.seal(session_id, binding=f"recap-id:{login}")
            serialized = self._cipher.seal(
                normalized, binding=f"recap-items:{login}:{session_id}"
            )
            connection.execute(
                "INSERT INTO today_recaps(login, session_id, created_at, items) VALUES (?, ?, ?, ?)",
                (login, encrypted_session_id, created_at, serialized),
            )
            connection.execute(
                "DELETE FROM today_recaps WHERE login = ? AND rowid NOT IN "
                "(SELECT rowid FROM today_recaps WHERE login = ? "
                "ORDER BY created_at DESC, rowid DESC LIMIT ?)",
                (login, login, self.recap_limit),
            )
        return self._recap_snapshot(login, encrypted_session_id, created_at, serialized)[0]

    def list_recaps(self, login: str, *, limit: int = 30) -> list[dict]:
        bounded_limit = max(1, min(int(limit), self.recap_limit, 100))
        login = self._normalize_login(login)
        with self._connect() as connection:
            rows = connection.execute(
                "SELECT session_id, created_at, items FROM today_recaps WHERE login = ? "
                "ORDER BY created_at DESC, rowid DESC LIMIT ?",
                (login, bounded_limit),
            ).fetchall()
            snapshots = []
            for row in rows:
                snapshot, legacy = self._recap_snapshot(login, *row)
                if legacy:
                    self._migrate_recap(connection, login, row[0], snapshot)
                snapshots.append(snapshot)
        return snapshots

    def _time_log_snapshot(self, login: str, row) -> tuple[dict, bool]:
        encrypted_session, encrypted_identity, payload_value, status_value = row
        if encrypted_session.startswith("v1:"):
            session_id = self._cipher.open(
                encrypted_session, binding=f"time-log-session:{login}"
            )[0]
            identity = self._cipher.open(
                encrypted_identity, binding=f"time-log-identity:{login}:{session_id}"
            )[0]
            payload, legacy = self._cipher.open(
                str(payload_value), binding=f"time-log-payload:{login}:{session_id}:{identity}"
            )
            if not isinstance(session_id, str) or not isinstance(identity, str) or not isinstance(payload, dict):
                raise PrivateStateEncryptionError("private state could not be decrypted")
            return {
                "session_id": session_id, "identity": identity,
                "actual_minutes": payload.get("actual_minutes"), "status": payload.get("status"),
                "stored_session": encrypted_session, "stored_identity": encrypted_identity,
            }, legacy
        return {
            "session_id": encrypted_session, "identity": encrypted_identity,
            "actual_minutes": payload_value, "status": status_value,
            "stored_session": encrypted_session, "stored_identity": encrypted_identity,
        }, True

    def _sealed_time_log(self, login: str, session_id: str, identity: str, actual: int, status: str) -> tuple[str, str, str]:
        return (
            self._cipher.seal(session_id, binding=f"time-log-session:{login}"),
            self._cipher.seal(identity, binding=f"time-log-identity:{login}:{session_id}"),
            self._cipher.seal(
                {"actual_minutes": actual, "status": status},
                binding=f"time-log-payload:{login}:{session_id}:{identity}",
            ),
        )

    def _migrate_time_log(self, connection, login: str, snapshot: dict) -> dict:
        sealed_session, sealed_identity, sealed_payload = self._sealed_time_log(
            login, snapshot["session_id"], snapshot["identity"],
            snapshot["actual_minutes"], snapshot["status"],
        )
        connection.execute(
            "UPDATE today_time_logs SET session_id = ?, identity = ?, actual_minutes = ?, status = 'sealed' "
            "WHERE login = ? AND session_id = ? AND identity = ?",
            (sealed_session, sealed_identity, sealed_payload, login,
             snapshot["stored_session"], snapshot["stored_identity"]),
        )
        return {**snapshot, "stored_session": sealed_session, "stored_identity": sealed_identity}

    def begin_time_log(self, login: str, session_id: str, identity: str, actual_minutes: int) -> str:
        """Claim one recap item for upstream logging, returning its current state."""
        login = self._normalize_login(login)
        with self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            rows = connection.execute(
                "SELECT session_id, identity, actual_minutes, status FROM today_time_logs WHERE login = ?",
                (login,),
            ).fetchall()
            match = None
            for row in rows:
                candidate, legacy = self._time_log_snapshot(login, row)
                if candidate["session_id"] == session_id and candidate["identity"] == identity:
                    match = self._migrate_time_log(connection, login, candidate) if legacy else candidate
                    break
            if match is None:
                sealed_session, sealed_identity, sealed_payload = self._sealed_time_log(
                    login, session_id, identity, actual_minutes, "pending"
                )
                connection.execute(
                    "INSERT INTO today_time_logs(login, session_id, identity, actual_minutes, status) "
                    "VALUES (?, ?, ?, ?, 'sealed')",
                    (login, sealed_session, sealed_identity, sealed_payload),
                )
                return "claimed"
            if match["actual_minutes"] != actual_minutes:
                raise ValueError("logged recap time cannot be changed")
            if match["status"] == "failed":
                sealed_payload = self._sealed_time_log(
                    login, session_id, identity, actual_minutes, "pending"
                )[2]
                connection.execute(
                    "UPDATE today_time_logs SET actual_minutes = ?, status = 'sealed' "
                    "WHERE login = ? AND session_id = ? AND identity = ?",
                    (sealed_payload, login, match["stored_session"], match["stored_identity"]),
                )
                return "claimed"
            return match["status"]

    def finish_time_log(self, login: str, session_id: str, identity: str, *, succeeded: bool) -> None:
        login = self._normalize_login(login)
        with self._connect() as connection:
            rows = connection.execute(
                "SELECT session_id, identity, actual_minutes, status FROM today_time_logs WHERE login = ?",
                (login,),
            ).fetchall()
            for row in rows:
                match, legacy = self._time_log_snapshot(login, row)
                if match["session_id"] != session_id or match["identity"] != identity or match["status"] != "pending":
                    continue
                if legacy:
                    match = self._migrate_time_log(connection, login, match)
                sealed_payload = self._sealed_time_log(
                    login, session_id, identity, match["actual_minutes"],
                    "logged" if succeeded else "failed",
                )[2]
                connection.execute(
                    "UPDATE today_time_logs SET actual_minutes = ?, status = 'sealed' "
                    "WHERE login = ? AND session_id = ? AND identity = ?",
                    (sealed_payload, login, match["stored_session"], match["stored_identity"]),
                )
                return

    def apply(
        self,
        login: str,
        operation_id: str,
        action: str,
        item_id: str,
        *,
        direction: str | None = None,
    ) -> dict:
        login = self._normalize_login(login)
        if not operation_id or not item_id:
            raise ValueError("operation_id and item_id are required")
        if action not in {"add", "remove", "move"}:
            raise ValueError("unsupported Today action")
        if action == "move" and direction not in {"up", "down"}:
            raise ValueError("move direction must be up or down")

        with self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            row = connection.execute(
                "SELECT revision, ids, capacity_minutes, estimates, plan_date, timezone "
                "FROM today_plans WHERE login = ?", (login,)
            ).fetchone()
            snapshot, _legacy = self._snapshot(row, login)
            duplicate = connection.execute(
                "SELECT 1 FROM today_operations WHERE login = ? AND operation_id = ?",
                (login, operation_id),
            ).fetchone()
            if duplicate:
                return snapshot

            ids = list(snapshot["ids"])
            estimates = dict(snapshot["estimates"])
            changed = False
            if action == "add":
                if item_id not in ids:
                    if len(ids) >= self.limit:
                        raise TodayPlanFull("Today is limited to five items")
                    ids.append(item_id)
                    changed = True
            elif action == "remove":
                if item_id in ids:
                    ids.remove(item_id)
                    estimates.pop(item_id, None)
                    changed = True
            else:
                try:
                    index = ids.index(item_id)
                except ValueError:
                    index = -1
                target = index - 1 if direction == "up" else index + 1
                if index >= 0 and 0 <= target < len(ids):
                    ids[index], ids[target] = ids[target], ids[index]
                    changed = True

            revision = snapshot["revision"] + (1 if changed else 0)
            sealed = self._sealed_plan(login, {
                "ids": ids, "capacity_minutes": snapshot["capacity_minutes"],
                "estimates": estimates, "plan_date": snapshot.get("plan_date"),
                "timezone": snapshot.get("timezone"),
                "first_task_state": snapshot.get("first_task_state", ""),
            })
            if row is None:
                connection.execute(
                    "INSERT INTO today_plans(login, revision, ids, capacity_minutes, estimates, plan_date, timezone) "
                    "VALUES (?, ?, ?, ?, ?, ?, ?)",
                    (login, revision, sealed, None, "{}", None, None),
                )
            elif changed:
                connection.execute(
                    "UPDATE today_plans SET revision = ?, ids = ?, capacity_minutes = NULL, "
                    "estimates = '{}', plan_date = NULL, timezone = NULL WHERE login = ?",
                    (revision, sealed, login),
                )
            self._record_operation(connection, login, operation_id)
            result = {
                "revision": revision,
                "ids": ids,
                "capacity_minutes": snapshot["capacity_minutes"],
                "estimates": estimates,
            }
            if snapshot.get("plan_date"):
                result["plan_date"] = snapshot["plan_date"]
                result["timezone"] = snapshot["timezone"]
            if snapshot.get("first_task_state"):
                result["first_task_state"] = snapshot["first_task_state"]
            return result

    def apply_batch(self, login: str, operations: list[dict]) -> dict:
        """Apply an ordered batch with one lock and receipt per operation."""
        login = self._normalize_login(login)
        with self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            row = connection.execute(
                "SELECT revision, ids, capacity_minutes, estimates, plan_date, timezone "
                "FROM today_plans WHERE login = ?", (login,)
            ).fetchone()
            snapshot, _legacy = self._snapshot(row, login)
            ids = list(snapshot["ids"])
            capacity_minutes = snapshot["capacity_minutes"]
            estimates = dict(snapshot["estimates"])
            plan_date = snapshot.get("plan_date")
            timezone = snapshot.get("timezone")
            first_task_state = snapshot.get("first_task_state", "")
            revision = snapshot["revision"]
            accepted: list[str] = []
            duplicates: list[str] = []
            rejected: list[dict[str, str]] = []

            for operation in operations:
                operation_id = operation.get("operation_id", "")
                action = operation.get("action", "")
                item_id = operation.get("item_id", "")
                direction = operation.get("direction")
                if not operation_id or not item_id:
                    raise ValueError("operation_id and item_id are required")
                if action not in {"add", "remove", "move", "configure", "rollover", "activate"}:
                    raise ValueError("unsupported Today action")
                if action == "move" and direction not in {"up", "down"}:
                    raise ValueError("move direction must be up or down")
                duplicate = connection.execute(
                    "SELECT 1 FROM today_operations WHERE login = ? AND operation_id = ?",
                    (login, operation_id),
                ).fetchone()
                if duplicate:
                    duplicates.append(operation_id)
                    continue

                base_revision = operation.get("base_revision")
                if base_revision is None:
                    base_revision = snapshot["revision"]
                if not isinstance(base_revision, int) or isinstance(base_revision, bool) or base_revision < 0:
                    raise ValueError("base_revision must be a non-negative integer")
                if action != "activate" and base_revision < snapshot["revision"]:
                    self._record_operation(connection, login, operation_id)
                    rejected.append({"operation_id": operation_id, "reason": "stale_intent"})
                    continue

                changed = False
                if action == "add":
                    if item_id not in ids:
                        if len(ids) >= self.limit:
                            rejected.append({"operation_id": operation_id, "reason": "today_full"})
                            continue
                        ids.append(item_id)
                        changed = True
                elif action == "remove":
                    if item_id in ids:
                        ids.remove(item_id)
                        estimates.pop(item_id, None)
                        changed = True
                elif action == "move":
                    try:
                        index = ids.index(item_id)
                    except ValueError:
                        index = -1
                    target = index - 1 if direction == "up" else index + 1
                    if index >= 0 and 0 <= target < len(ids):
                        ids[index], ids[target] = ids[target], ids[index]
                        changed = True
                elif action == "configure":
                    proposed_capacity = operation.get("capacity_minutes")
                    if proposed_capacity is not None and (
                        not isinstance(proposed_capacity, int) or isinstance(proposed_capacity, bool)
                        or proposed_capacity < 15 or proposed_capacity > 1440
                    ):
                        raise ValueError("capacity_minutes must be between 15 and 1440")
                    proposed_estimates = operation.get("estimates", {})
                    if not isinstance(proposed_estimates, dict):
                        raise ValueError("estimates must be an object")
                    normalized_estimates = {}
                    for estimate_id, minutes in proposed_estimates.items():
                        if estimate_id not in ids:
                            continue
                        if not isinstance(minutes, int) or isinstance(minutes, bool) or minutes < 5 or minutes > 1440:
                            raise ValueError("estimate minutes must be between 5 and 1440")
                        normalized_estimates[estimate_id] = minutes
                    changed = capacity_minutes != proposed_capacity or estimates != normalized_estimates
                    capacity_minutes = proposed_capacity
                    estimates = normalized_estimates
                elif action == "activate":
                    proposed_state = operation.get("activation_state")
                    if item_id != "first-task" or proposed_state not in {"coaching", "complete"}:
                        raise ValueError("first-task activation state is invalid")
                    rank = {"": 0, "coaching": 1, "complete": 2}
                    changed = rank[proposed_state] > rank[first_task_state]
                    if changed:
                        first_task_state = proposed_state
                else:
                    proposed_date = operation.get("plan_date")
                    proposed_timezone = operation.get("timezone")
                    try:
                        if date.fromisoformat(proposed_date or "").isoformat() != proposed_date:
                            raise ValueError
                    except (TypeError, ValueError):
                        raise ValueError("plan_date must be an ISO calendar date") from None
                    if not isinstance(proposed_timezone, str) or not proposed_timezone.strip() or len(proposed_timezone) > 100:
                        raise ValueError("timezone is required and bounded")
                    proposed_ids = operation.get("ids")
                    if not isinstance(proposed_ids, list) or len(proposed_ids) > self.limit or any(
                        not isinstance(candidate, str) or not candidate or len(candidate) > 500
                        for candidate in proposed_ids
                    ):
                        raise ValueError("rollover IDs are invalid or exceed the Today limit")
                    if len(set(proposed_ids)) != len(proposed_ids):
                        raise ValueError("rollover IDs must be unique")
                    proposed_capacity = operation.get("capacity_minutes")
                    if proposed_capacity is not None and (
                        not isinstance(proposed_capacity, int) or isinstance(proposed_capacity, bool)
                        or proposed_capacity < 15 or proposed_capacity > 1440
                    ):
                        raise ValueError("capacity_minutes must be between 15 and 1440")
                    proposed_estimates = operation.get("estimates", {})
                    if not isinstance(proposed_estimates, dict):
                        raise ValueError("estimates must be an object")
                    normalized_estimates = {}
                    for estimate_id, minutes in proposed_estimates.items():
                        if estimate_id not in proposed_ids:
                            continue
                        if not isinstance(minutes, int) or isinstance(minutes, bool) or minutes < 5 or minutes > 1440:
                            raise ValueError("estimate minutes must be between 5 and 1440")
                        normalized_estimates[estimate_id] = minutes
                    changed = (ids != proposed_ids or capacity_minutes != proposed_capacity or
                               estimates != normalized_estimates or plan_date != proposed_date or
                               timezone != proposed_timezone.strip())
                    ids = list(proposed_ids)
                    capacity_minutes = proposed_capacity
                    estimates = normalized_estimates
                    plan_date = proposed_date
                    timezone = proposed_timezone.strip()

                revision += 1 if changed else 0
                self._record_operation(connection, login, operation_id)
                accepted.append(operation_id)

            sealed = self._sealed_plan(login, {
                "ids": ids, "capacity_minutes": capacity_minutes, "estimates": estimates,
                "plan_date": plan_date, "timezone": timezone,
                "first_task_state": first_task_state,
            })
            if row is None:
                connection.execute(
                    "INSERT INTO today_plans(login, revision, ids, capacity_minutes, estimates, plan_date, timezone) "
                    "VALUES (?, ?, ?, ?, ?, ?, ?)",
                    (login, revision, sealed, None, "{}", None, None),
                )
            elif accepted:
                connection.execute(
                    "UPDATE today_plans SET revision = ?, ids = ?, capacity_minutes = ?, estimates = ?, "
                    "plan_date = ?, timezone = ? WHERE login = ?",
                    (revision, sealed, None, "{}", None, None, login),
                )
            result = {
                "revision": revision,
                "ids": ids,
                "capacity_minutes": capacity_minutes,
                "estimates": estimates,
                "accepted_operation_ids": accepted,
                "duplicate_operation_ids": duplicates,
                "rejected_operations": rejected,
            }
            if plan_date:
                result["plan_date"] = plan_date
                result["timezone"] = timezone
            if first_task_state:
                result["first_task_state"] = first_task_state
            return result
