"""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_config 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_config(), 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] >= 7: 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( "CREATE TABLE IF NOT EXISTS week_reschedules (" "login TEXT NOT NULL, operation_id TEXT NOT NULL, result TEXT NOT NULL, " "created_at REAL NOT NULL, PRIMARY KEY (login, operation_id))" ) connection.execute( "CREATE TABLE IF NOT EXISTS week_item_pulls (" "login TEXT NOT NULL, operation_id TEXT NOT NULL, result TEXT NOT NULL, " "created_at REAL NOT NULL, PRIMARY KEY (login, operation_id))" ) connection.execute("PRAGMA user_version = 7") 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:", "v2:")): 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_start_day_plan(self, login: str) -> dict: """Return the next private planned day without changing either plan.""" login = self._normalize_login(login) with self._connect() as connection: week_row = connection.execute( "SELECT revision, payload FROM week_plans WHERE login = ?", (login,) ).fetchone() if week_row is not None: week = self._week_snapshot(week_row, login) local_date = datetime.fromtimestamp( self.clock(), ZoneInfo(week["timezone"]) ).date().isoformat() for day in week["days"]: if ( isinstance(day, dict) and isinstance(day.get("ids"), list) and day["ids"] and day.get("plan_date", "") <= local_date ): return {**day, "timezone": week.get("timezone")} tomorrow_row = connection.execute( "SELECT revision, payload FROM tomorrow_plans WHERE login = ?", (login,) ).fetchone() tomorrow = self._tomorrow_snapshot(tomorrow_row, login) return tomorrow if tomorrow.get("ids") and tomorrow.get("plan_date") else {} 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, availability_defaults: list[int] | None = None, ) -> 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() seen_ids = 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, ) if "free_windows" in day: windows = day["free_windows"] if not isinstance(windows, list) or len(windows) > 16: raise ValueError("free windows must be a bounded list") normalized_windows = [] previous_end = -1 for window in windows: if not isinstance(window, dict) or set(window) != {"start_time", "end_time"}: raise ValueError("free windows must contain only start and end times") clock_values = [] for key in ("start_time", "end_time"): value = window[key] if ( not isinstance(value, str) or len(value) != 5 or value[2] != ":" or not value[:2].isdigit() or not value[3:].isdigit() or int(value[:2]) > 23 or int(value[3:]) > 59 ): raise ValueError("free windows must use HH:mm times") clock_values.append(int(value[:2]) * 60 + int(value[3:])) start, end = clock_values if start < previous_end or end <= start: raise ValueError("free windows must be ordered and non-overlapping") normalized_windows.append({ "start_time": window["start_time"], "end_time": window["end_time"], }) previous_end = end plan["free_windows"] = normalized_windows if "start_times" in day: start_times = day["start_times"] if not isinstance(start_times, dict) or len(start_times) > 5: raise ValueError("task start times must be a bounded object") if not set(start_times).issubset(plan["ids"]): raise ValueError("task start times must belong to work on the same date") occupied = [] for identity in plan["ids"]: if identity not in start_times: continue value = start_times[identity] if ( not isinstance(value, str) or len(value) != 5 or value[2] != ":" or not value[:2].isdigit() or not value[3:].isdigit() or int(value[:2]) > 23 or int(value[3:]) > 59 ): raise ValueError("task start times must use HH:mm times") start = int(value[:2]) * 60 + int(value[3:]) minutes = plan["estimates"].get(identity) if not isinstance(minutes, int) or start + minutes > 1440: raise ValueError("task start times must fit their estimates within the date") end = start + minutes if any(start < other_end and end > other_start for other_start, other_end in occupied): raise ValueError("task start times must not overlap") windows = plan.get("free_windows") if windows and not any( int(window["start_time"][:2]) * 60 + int(window["start_time"][3:]) <= start and end <= int(window["end_time"][:2]) * 60 + int(window["end_time"][3:]) for window in windows ): raise ValueError("task start times must fit retained free windows") occupied.append((start, end)) plan["start_times"] = {identity: start_times[identity] for identity in plan["ids"] if identity in start_times} duplicate_ids = seen_ids.intersection(plan["ids"]) if duplicate_ids: raise ValueError("work must be assigned to only one Week Ahead date") seen_ids.update(plan["ids"]) plan.pop("timezone") normalized.append(plan) normalized.sort(key=lambda item: item["plan_date"]) result = {"timezone": timezone, "days": normalized} if availability_defaults is not None: if ( not isinstance(availability_defaults, list) or len(availability_defaults) != 7 or any( not isinstance(value, int) or isinstance(value, bool) or value < 0 or value > 1440 for value in availability_defaults ) ): raise ValueError("availability defaults must contain seven weekday capacities") result["availability_defaults"] = list(availability_defaults) return result def replace_week( self, login: str, *, base_revision: int, days: list[dict], timezone: str, availability_defaults: list[int] | None = None, ) -> 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, availability_defaults=availability_defaults, ) 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 reschedule_today_to_week( self, login: str, *, operation_id: str, identity: str, estimate_minutes: int, plan_date: str, today_revision: int, week_revision: int, allow_over_capacity: bool = False, ) -> dict: """Atomically move one current Today item to one Week Ahead date.""" login = self._normalize_login(login) if not isinstance(operation_id, str) or not operation_id.strip() or len(operation_id) > 100: raise ValueError("operation_id is required and bounded") operation_id = operation_id.strip() if not isinstance(identity, str) or not identity or len(identity) > 500: raise ValueError("identity is required and bounded") if ( not isinstance(estimate_minutes, int) or isinstance(estimate_minutes, bool) or estimate_minutes < 5 or estimate_minutes > 1440 ): raise ValueError("estimate_minutes must be between 5 and 1440") with self._connect() as connection: connection.execute("BEGIN IMMEDIATE") receipt = connection.execute( "SELECT result FROM week_reschedules WHERE login = ? AND operation_id = ?", (login, operation_id), ).fetchone() if receipt: result, _legacy = self._cipher.open( receipt[0], binding=f"week-reschedule:{login}:{operation_id}" ) if not isinstance(result, dict): raise PrivateStateEncryptionError("private state could not be decrypted") return result 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) week_row = connection.execute( "SELECT revision, payload FROM week_plans WHERE login = ?", (login,) ).fetchone() week = self._week_snapshot(week_row, login) if today["revision"] != today_revision or identity not in today["ids"]: raise TodayPromotionConflict(today, week) if week["revision"] != week_revision: raise WeekPlanConflict(week) days = [] for source in week["days"]: estimates = dict(source.get("estimates", {})) estimates.pop(identity, None) days.append({ **source, "ids": [item_id for item_id in source.get("ids", []) if item_id != identity], "estimates": estimates, }) destination = next((day for day in days if day["plan_date"] == plan_date), None) if destination is None: destination = { "plan_date": plan_date, "ids": [], "capacity_minutes": None, "estimates": {}, } days.append(destination) destination["ids"].append(identity) destination["estimates"][identity] = estimate_minutes planned_minutes = sum( int(destination["estimates"].get(item_id, 0)) for item_id in destination["ids"] ) if ( destination.get("capacity_minutes") is not None and planned_minutes > destination["capacity_minutes"] and not allow_over_capacity ): raise ValueError("explicit overload confirmation is required") normalized_week = self._normalize_week( days=days, timezone=week["timezone"], availability_defaults=week.get("availability_defaults"), ) today_ids = [item_id for item_id in today["ids"] if item_id != identity] today_estimates = { item_id: minutes for item_id, minutes in today["estimates"].items() if item_id != identity } today_result = { **today, "revision": today_revision + 1, "ids": today_ids, "estimates": today_estimates, } week_result = {"revision": week_revision + 1, **normalized_week} connection.execute( "UPDATE today_plans SET revision = ?, ids = ?, capacity_minutes = NULL, " "estimates = '{}', plan_date = NULL, timezone = NULL WHERE login = ?", (today_result["revision"], self._sealed_plan(login, today_result), login), ) connection.execute( "UPDATE week_plans SET revision = ?, payload = ? WHERE login = ?", (week_result["revision"], self._cipher.seal( normalized_week, binding=f"week:{login}" ), login), ) result = {"today": today_result, "week": week_result} connection.execute( "INSERT INTO week_reschedules(login, operation_id, result, created_at) " "VALUES (?, ?, ?, ?)", (login, operation_id, self._cipher.seal( result, binding=f"week-reschedule:{login}:{operation_id}" ), self.clock()), ) return result def pull_week_item( self, login: str, *, operation_id: str, identity: str, today_revision: int, week_revision: int, allow_over_capacity: bool = False, ) -> dict: """Atomically append one Week Ahead item to Today and remove it from Week.""" login = self._normalize_login(login) if not isinstance(operation_id, str) or not operation_id.strip() or len(operation_id) > 100: raise ValueError("operation_id is required and bounded") if not isinstance(identity, str) or not identity or len(identity) > 500: raise ValueError("identity is required and bounded") operation_id = operation_id.strip() with self._connect() as connection: connection.execute("BEGIN IMMEDIATE") receipt = connection.execute( "SELECT result FROM week_item_pulls WHERE login = ? AND operation_id = ?", (login, operation_id), ).fetchone() if receipt: result, _legacy = self._cipher.open( receipt[0], binding=f"week-item-pull:{login}:{operation_id}" ) if not isinstance(result, dict): raise PrivateStateEncryptionError("private state could not be decrypted") return result 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) week_row = connection.execute( "SELECT revision, payload FROM week_plans WHERE login = ?", (login,), ).fetchone() week = self._week_snapshot(week_row, login) if today["revision"] != today_revision: raise TodayPromotionConflict(today, week) if week["revision"] != week_revision: raise WeekPlanConflict(week) if identity in today["ids"]: raise ValueError("work is already in Today") if len(today["ids"]) >= self.limit: raise TodayPlanFull("Today plan is full") source = next((day for day in week["days"] if identity in day["ids"]), None) if source is None: raise WeekPlanConflict(week) estimate = source.get("estimates", {}).get(identity) planned_minutes = sum( int(today["estimates"].get(item_id, 0)) for item_id in today["ids"] ) + (int(estimate) if estimate else 0) if ( today.get("capacity_minutes") is not None and planned_minutes > today["capacity_minutes"] and not allow_over_capacity ): raise ValueError("moving work over Today capacity requires explicit overload confirmation") today_result = { **today, "revision": today_revision + 1, "ids": [*today["ids"], identity], "estimates": {**today["estimates"], **({identity: estimate} if estimate else {})}, } remaining_days = [] for day in week["days"]: estimates = dict(day.get("estimates", {})) estimates.pop(identity, None) remaining_days.append({ **day, "ids": [item_id for item_id in day["ids"] if item_id != identity], "estimates": estimates, }) week_payload = {"timezone": week["timezone"], "days": remaining_days} week_result = {"revision": week_revision + 1, **week_payload} 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, today_result["revision"], self._sealed_plan(login, today_result)), ) connection.execute( "UPDATE week_plans SET revision = ?, payload = ? WHERE login = ?", (week_result["revision"], self._cipher.seal(week_payload, binding=f"week:{login}"), login), ) result = {"today": today_result, "week": week_result} connection.execute( "INSERT INTO week_item_pulls(login, operation_id, result, created_at) VALUES (?, ?, ?, ?)", (login, operation_id, self._cipher.seal( result, binding=f"week-item-pull:{login}:{operation_id}" ), self.clock()), ) return result def promote_week( self, login: str, *, promotion_id: str, week_revision: int, plan_date: str, today_revision: int, allow_future: bool = False, ) -> 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 and not allow_future: 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 def reconcile_week( self, login: str, *, promotion_id: str, week_revision: int, plan_date: str, today_revision: int, ids: list[str], capacity_minutes: int | None, estimates: dict[str, int], ) -> dict: """Atomically replace Today from preserved Today/due-week work and consume that day.""" 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, week) allowed = set(today["ids"]) | set(day["ids"]) if any(item_id not in allowed for item_id in ids): raise ValueError("selected work must come from Today or the due Week Ahead day") normalized = self._normalize_tomorrow( ids=ids, capacity_minutes=capacity_minutes, estimates=estimates, plan_date=plan_date, timezone=week["timezone"], ) today_result = {"revision": today_revision + 1, **normalized} 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, today_result["revision"], self._sealed_plan(login, today_result)), ) remaining_payload = {"timezone": week["timezone"], "days": [ item for item in week["days"] if item["plan_date"] != plan_date ]} week_result = {"revision": week_revision + 1, **remaining_payload} connection.execute( "UPDATE week_plans SET revision = ?, payload = ? WHERE login = ?", (week_result["revision"], self._cipher.seal( remaining_payload, binding=f"week:{login}" ), login), ) result = {"today": today_result, "week": week_result} 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, "entries": [], "running": False, "break_deadline_at": None, "updated_at": None, } @staticmethod def _normalize_session_entries(entries: list[dict]) -> list[dict]: if not isinstance(entries, list) or len(entries) > 20: raise ValueError("Today session entries are limited to 20 items") normalized = [] seen = set() for item in entries: identity = item.get("identity", "") if isinstance(item, dict) else "" elapsed_ms = item.get("elapsed_ms") if isinstance(item, dict) else None if not isinstance(identity, str) or not identity.strip() or len(identity) > 500: raise ValueError("Today session entry identity is required and bounded") identity = identity.strip() if identity in seen: raise ValueError("Today session entry identities must be unique") if (not isinstance(elapsed_ms, int) or isinstance(elapsed_ms, bool) or elapsed_ms < 0 or elapsed_ms > 7 * 24 * 60 * 60 * 1000): raise ValueError("Today session elapsed time is invalid") seen.add(identity) normalized.append({"identity": identity, "elapsed_ms": elapsed_ms}) return normalized def _session_snapshot(self, row, login: str) -> tuple[dict, bool]: if row is None: return self._empty_session(), False if row[1].startswith(("v1:", "v2:")): payload, legacy = self._cipher.open(row[1], binding=f"session:{login}") if not isinstance(payload, dict): raise PrivateStateEncryptionError("private state could not be decrypted") if "entries" not in payload: payload["entries"] = ([{ "identity": payload.get("identity", ""), "elapsed_ms": int(payload.get("elapsed_ms", 0)), }] if payload.get("identity") else []) legacy = True return {"revision": int(row[0]), **payload, "updated_at": row[6]}, legacy session = { "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], } session["entries"] = ([{ "identity": session["identity"], "elapsed_ms": session["elapsed_ms"], }] if session["identity"] else []) return session, 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"], "entries": session["entries"], "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, entries: list[dict] | 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 normalized_entries = self._normalize_session_entries( entries if entries is not None else ([{"identity": identity, "elapsed_ms": elapsed_ms}] if identity else []) ) active_entries = [entry for entry in normalized_entries if entry["identity"] == identity] if identity and (len(active_entries) != 1 or active_entries[0]["elapsed_ms"] != elapsed_ms): raise ValueError("active Today session must match its ledger entry") session = { "revision": revision, "device_id": device_id, "identity": identity, "elapsed_ms": elapsed_ms, "entries": normalized_entries, "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:", "v2:")): 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:", "v2:")): 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