from __future__ import annotations

from contextlib import contextmanager
from dataclasses import dataclass
from datetime import date, datetime, time, timedelta, timezone
import hashlib
import json
import os
from pathlib import Path
import re
import secrets
import sqlite3
import threading
from typing import Literal
import unicodedata
import uuid

from pydantic import BaseModel, Field

from app.core.config import get_settings
from app.core.mock_state import MODULE_CATALOG
from app.models.tenant import Tenant
from app.models.tenant_module import TenantModule
from app.models.user import User
from app.models.venue import Venue

SUPER_ADMIN_TENANT_ID = "tenant_powerup_admin"
SUPER_ADMIN_VENUE_ID = "venue_powerup_admin"
SUPER_ADMIN_USER_ID = "user_powerup_admin"
SUPER_ADMIN_TENANT_NAME = "PowerUp Admin"
SUPER_ADMIN_TENANT_SLUG = "powerup-admin"
SUPER_ADMIN_USERNAME = "zanchinfede@gmail.com"
SUPER_ADMIN_EMAIL = "superadmin@powerup.local"
TENANT_PERMISSION_KEYS = (
    "assistant",
    "ordini",
    "prenotazioni",
    "documents",
    "menu",
    "homemade",
    "fiscal_documents",
    "timeclock",
    "inventory",
)
DEFAULT_STAFF_PERMISSIONS = (
    "prenotazioni",
    "menu",
    "homemade",
)


def _utcnow() -> datetime:
    return datetime.now(timezone.utc)


def _iso_now() -> str:
    return _utcnow().isoformat()


def _normalize_slug(value: str) -> str:
    normalized = unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode("ascii")
    slug = re.sub(r"[^a-zA-Z0-9]+", "-", normalized.lower()).strip("-")
    return slug or f"tenant-{secrets.token_hex(3)}"


def _normalize_lookup(value: str) -> str:
    return value.strip().lower()


def _hash_password(password: str) -> str:
    salt = os.urandom(16)
    iterations = 240_000
    digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, iterations)
    return f"pbkdf2_sha256${iterations}${salt.hex()}${digest.hex()}"


def _verify_password(password: str, stored_hash: str) -> bool:
    try:
        algorithm, raw_iterations, salt_hex, digest_hex = stored_hash.split("$", 3)
    except ValueError:
        return False

    if algorithm != "pbkdf2_sha256":
        return False

    try:
        iterations = int(raw_iterations)
        salt = bytes.fromhex(salt_hex)
        expected_digest = bytes.fromhex(digest_hex)
    except ValueError:
        return False

    candidate = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, iterations)
    return secrets.compare_digest(candidate, expected_digest)


def _normalize_permission_values(values: list[str] | tuple[str, ...] | None) -> tuple[str, ...]:
    if not values:
        return tuple()

    normalized: list[str] = []
    seen: set[str] = set()
    for raw_value in values:
        value = str(raw_value or "").strip().lower()
        if value not in TENANT_PERMISSION_KEYS or value in seen:
            continue
        normalized.append(value)
        seen.add(value)
    return tuple(normalized)


def _resolve_role_permissions(role: str, permissions_json: str | None) -> tuple[str, ...]:
    if role in {"owner", "super_admin"}:
        return TENANT_PERMISSION_KEYS

    if permissions_json is None:
        return DEFAULT_STAFF_PERMISSIONS if role == "staff" else tuple()

    raw_permissions: list[str] | tuple[str, ...] | None = None
    try:
        parsed = json.loads(permissions_json)
    except json.JSONDecodeError:
        parsed = None
    if isinstance(parsed, list):
        raw_permissions = [str(item) for item in parsed]

    normalized = _normalize_permission_values(raw_permissions)
    return normalized


def _serialize_role_permissions(role: str, permissions: list[str] | tuple[str, ...] | None) -> str | None:
    if role in {"owner", "super_admin"}:
        return None
    return json.dumps(list(_normalize_permission_values(permissions)))


class RegisterTenantPayload(BaseModel):
    locale_name: str = Field(min_length=2)
    address: str = ""
    phone_number: str | None = None
    whatsapp_number: str | None = None
    admin_name: str = Field(min_length=2)
    username: str = Field(min_length=3)
    email: str = Field(min_length=5)
    password: str = Field(min_length=6)


class LoginPayload(BaseModel):
    identifier: str = Field(min_length=3)
    password: str = Field(min_length=1)


class BootstrapSuperAdminPayload(BaseModel):
    password: str = Field(min_length=8)


class UpdatePasswordPayload(BaseModel):
    current_password: str | None = Field(default=None, min_length=1)
    new_password: str = Field(min_length=8)


class AdminCreateTenantPayload(RegisterTenantPayload):
    pass


class AdminUpdateTenantAdminPayload(BaseModel):
    locale_name: str = Field(min_length=2)
    address: str = ""
    phone_number: str | None = None
    whatsapp_number: str | None = None
    admin_name: str = Field(min_length=2)
    username: str = Field(min_length=3)
    email: str = Field(min_length=5)
    password: str | None = Field(default=None, min_length=8)


class TenantStaffUserCreatePayload(BaseModel):
    name: str = Field(min_length=2)
    username: str = Field(min_length=3)
    email: str = Field(min_length=5)
    phone_number: str | None = None
    password: str = Field(min_length=8)
    permissions: list[str] = Field(default_factory=list)


class TenantStaffUserUpdatePayload(BaseModel):
    name: str = Field(min_length=2)
    username: str = Field(min_length=3)
    email: str = Field(min_length=5)
    phone_number: str | None = None
    password: str | None = Field(default=None, min_length=8)
    permissions: list[str] = Field(default_factory=list)


class TimeclockOverviewQuery(BaseModel):
    user_id: str | None = None
    start_date: str | None = None
    end_date: str | None = None
    limit: int = Field(default=200, ge=1, le=1000)


class InventoryWarehouseCreatePayload(BaseModel):
    name: str = Field(min_length=2, max_length=120)


class InventoryStockWritePayload(BaseModel):
    product_id: int = Field(ge=1)
    quantity: float = Field(ge=0)
    operation: Literal["add", "set"] = "add"


class InventorySessionCreatePayload(BaseModel):
    inventory_date: str | None = None


class InventorySnapshotCreatePayload(BaseModel):
    snapshot_date: str | None = None


@dataclass(frozen=True)
class SessionIdentity:
    token: str
    tenant_id: str
    tenant_slug: str
    tenant_name: str
    home_tenant_id: str
    home_tenant_slug: str
    home_tenant_name: str
    user_id: str
    user_email: str
    username: str | None
    user_name: str | None
    role: str
    permissions: tuple[str, ...]
    database_path: str
    expires_at: str


@dataclass(frozen=True)
class MenuAssetRecord:
    id: str
    tenant_id: str
    original_name: str
    display_name: str
    mime_type: str
    kind: str
    file_size_bytes: int
    storage_path: str
    extracted_text: str
    analysis_text: str
    status: str
    error_detail: str | None
    created_at: str
    updated_at: str


@dataclass(frozen=True)
class FiscalDocumentRecord:
    id: str
    tenant_id: str
    original_name: str
    display_name: str
    mime_type: str
    kind: str
    file_size_bytes: int
    file_hash: str | None
    storage_path: str
    document_type: str
    document_number: str | None
    document_date: str | None
    supplier_name: str | None
    total_amount: float | None
    currency: str
    summary_text: str
    extracted_text: str
    status: str
    matching_status: str
    review_status: str
    error_detail: str | None
    created_at: str
    updated_at: str


@dataclass(frozen=True)
class FiscalDocumentSettingsRecord:
    tenant_id: str
    inbound_email: str | None
    updated_at: str | None


@dataclass(frozen=True)
class FiscalDocumentInboxItemRecord:
    id: str
    tenant_id: str
    message_id: str
    attachment_id: str
    file_hash: str | None
    subject: str
    sender: str | None
    received_at: str | None
    attachment_name: str
    mime_type: str
    sync_status: str
    document_id: str | None
    error_detail: str | None
    created_at: str
    updated_at: str


@dataclass(frozen=True)
class TimeclockEntryRecord:
    id: str
    user_id: str
    user_name: str | None
    username: str | None
    user_email: str | None
    started_at: str
    ended_at: str | None
    duration_seconds: int | None
    started_source: str
    ended_source: str | None
    notes: str | None
    created_at: str
    updated_at: str


@dataclass(frozen=True)
class AssistantThreadRecord:
    id: str
    tenant_id: str
    user_id: str
    surface: str
    status: str
    state_json: str | None
    created_at: str
    updated_at: str


@dataclass(frozen=True)
class AssistantStoredMessage:
    id: str
    thread_id: str
    role: str
    content: str
    created_at: str
    sort_index: int


@dataclass(frozen=True)
class AssistantRunRecord:
    id: str
    thread_id: str
    tenant_id: str
    user_id: str
    surface: str
    route: str
    model: str
    user_message: str
    assistant_reply: str
    trace_json: str
    created_at: str


class TenantStore:
    def __init__(self) -> None:
        settings = get_settings()
        self._registry_path = Path(settings.tenancy_registry_database)
        self._tenant_dir = Path(settings.tenancy_databases_dir)
        self._menu_assets_dir = Path(settings.menu_asset_storage_dir)
        self._fiscal_documents_dir = Path(settings.fiscal_document_storage_dir)
        self._session_duration = timedelta(hours=settings.tenancy_session_duration_hours)
        self._lock = threading.Lock()
        self._ensure_storage()

    def _ensure_storage(self) -> None:
        self._registry_path.parent.mkdir(parents=True, exist_ok=True)
        self._tenant_dir.mkdir(parents=True, exist_ok=True)
        self._menu_assets_dir.mkdir(parents=True, exist_ok=True)
        self._fiscal_documents_dir.mkdir(parents=True, exist_ok=True)

        with self._connect_registry() as connection:
            connection.executescript(
                """
                CREATE TABLE IF NOT EXISTS tenants (
                    id TEXT PRIMARY KEY,
                    name TEXT NOT NULL,
                    slug TEXT NOT NULL UNIQUE,
                    database_path TEXT NOT NULL,
                    created_at TEXT NOT NULL
                );

                CREATE TABLE IF NOT EXISTS venues (
                    id TEXT PRIMARY KEY,
                    tenant_id TEXT NOT NULL,
                    name TEXT NOT NULL,
                    address TEXT NOT NULL DEFAULT '',
                    phone_number TEXT,
                    whatsapp_number TEXT,
                    created_at TEXT NOT NULL,
                    FOREIGN KEY(tenant_id) REFERENCES tenants(id)
                );

                CREATE TABLE IF NOT EXISTS users (
                    id TEXT PRIMARY KEY,
                    tenant_id TEXT NOT NULL,
                    name TEXT NOT NULL,
                    username TEXT NOT NULL UNIQUE,
                    email TEXT NOT NULL UNIQUE,
                    phone_number TEXT,
                    password_hash TEXT NOT NULL,
                    role TEXT NOT NULL,
                    permissions_json TEXT,
                    created_at TEXT NOT NULL,
                    FOREIGN KEY(tenant_id) REFERENCES tenants(id)
                );

                CREATE TABLE IF NOT EXISTS tenant_modules (
                    id TEXT PRIMARY KEY,
                    tenant_id TEXT NOT NULL,
                    module_key TEXT NOT NULL,
                    enabled INTEGER NOT NULL,
                    plan_name TEXT NOT NULL,
                    activated_at TEXT,
                    expires_at TEXT,
                    FOREIGN KEY(tenant_id) REFERENCES tenants(id)
                );

                CREATE UNIQUE INDEX IF NOT EXISTS idx_tenant_modules_unique
                    ON tenant_modules(tenant_id, module_key);

                CREATE TABLE IF NOT EXISTS sessions (
                    token TEXT PRIMARY KEY,
                    tenant_id TEXT NOT NULL,
                    user_id TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    expires_at TEXT NOT NULL,
                    FOREIGN KEY(tenant_id) REFERENCES tenants(id),
                    FOREIGN KEY(user_id) REFERENCES users(id)
                );

                CREATE TABLE IF NOT EXISTS tenant_llm_settings (
                    tenant_id TEXT NOT NULL,
                    scope TEXT NOT NULL,
                    config_json TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    PRIMARY KEY (tenant_id, scope),
                    FOREIGN KEY(tenant_id) REFERENCES tenants(id)
                );

                CREATE TABLE IF NOT EXISTS tenant_menu_assets (
                    id TEXT PRIMARY KEY,
                    tenant_id TEXT NOT NULL,
                    original_name TEXT NOT NULL,
                    display_name TEXT NOT NULL,
                    mime_type TEXT NOT NULL,
                    kind TEXT NOT NULL,
                    file_size_bytes INTEGER NOT NULL,
                    storage_path TEXT NOT NULL,
                    extracted_text TEXT NOT NULL DEFAULT '',
                    analysis_text TEXT NOT NULL DEFAULT '',
                    status TEXT NOT NULL,
                    error_detail TEXT,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    FOREIGN KEY(tenant_id) REFERENCES tenants(id)
                );

                CREATE INDEX IF NOT EXISTS idx_tenant_menu_assets_tenant
                    ON tenant_menu_assets(tenant_id, created_at DESC);

                CREATE TABLE IF NOT EXISTS tenant_fiscal_documents (
                    id TEXT PRIMARY KEY,
                    tenant_id TEXT NOT NULL,
                    original_name TEXT NOT NULL,
                    display_name TEXT NOT NULL,
                    mime_type TEXT NOT NULL,
                    kind TEXT NOT NULL,
                    file_size_bytes INTEGER NOT NULL,
                    file_hash TEXT,
                    storage_path TEXT NOT NULL,
                    document_type TEXT NOT NULL,
                    document_number TEXT,
                    document_date TEXT,
                    supplier_name TEXT,
                    total_amount REAL,
                    currency TEXT NOT NULL DEFAULT 'EUR',
                    summary_text TEXT NOT NULL DEFAULT '',
                    extracted_text TEXT NOT NULL DEFAULT '',
                    status TEXT NOT NULL,
                    matching_status TEXT NOT NULL DEFAULT 'pending',
                    review_status TEXT NOT NULL DEFAULT 'to_review',
                    error_detail TEXT,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    FOREIGN KEY(tenant_id) REFERENCES tenants(id)
                );

                CREATE INDEX IF NOT EXISTS idx_tenant_fiscal_documents_tenant
                    ON tenant_fiscal_documents(tenant_id, created_at DESC);

                CREATE TABLE IF NOT EXISTS tenant_fiscal_document_settings (
                    tenant_id TEXT PRIMARY KEY,
                    inbound_email TEXT,
                    updated_at TEXT NOT NULL,
                    FOREIGN KEY(tenant_id) REFERENCES tenants(id)
                );

                CREATE TABLE IF NOT EXISTS tenant_fiscal_document_inbox_items (
                    id TEXT PRIMARY KEY,
                    tenant_id TEXT NOT NULL,
                    message_id TEXT NOT NULL,
                    attachment_id TEXT NOT NULL,
                    file_hash TEXT,
                    subject TEXT NOT NULL DEFAULT '',
                    sender TEXT,
                    received_at TEXT,
                    attachment_name TEXT NOT NULL,
                    mime_type TEXT NOT NULL,
                    sync_status TEXT NOT NULL,
                    document_id TEXT,
                    error_detail TEXT,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    FOREIGN KEY(tenant_id) REFERENCES tenants(id),
                    FOREIGN KEY(document_id) REFERENCES tenant_fiscal_documents(id),
                    UNIQUE(tenant_id, message_id, attachment_id)
                );

                CREATE INDEX IF NOT EXISTS idx_tenant_fiscal_document_inbox_items_tenant
                    ON tenant_fiscal_document_inbox_items(tenant_id, created_at DESC);

                CREATE TABLE IF NOT EXISTS assistant_threads (
                    id TEXT PRIMARY KEY,
                    tenant_id TEXT NOT NULL,
                    user_id TEXT NOT NULL,
                    surface TEXT NOT NULL,
                    status TEXT NOT NULL DEFAULT 'active',
                    state_json TEXT,
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL,
                    FOREIGN KEY(tenant_id) REFERENCES tenants(id),
                    FOREIGN KEY(user_id) REFERENCES users(id)
                );

                CREATE INDEX IF NOT EXISTS idx_assistant_threads_lookup
                    ON assistant_threads(tenant_id, user_id, surface, status, updated_at DESC);

                CREATE TABLE IF NOT EXISTS assistant_messages (
                    id TEXT PRIMARY KEY,
                    thread_id TEXT NOT NULL,
                    role TEXT NOT NULL,
                    content TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    sort_index INTEGER NOT NULL,
                    FOREIGN KEY(thread_id) REFERENCES assistant_threads(id) ON DELETE CASCADE
                );

                CREATE INDEX IF NOT EXISTS idx_assistant_messages_thread
                    ON assistant_messages(thread_id, sort_index ASC);

                CREATE TABLE IF NOT EXISTS assistant_runs (
                    id TEXT PRIMARY KEY,
                    thread_id TEXT NOT NULL,
                    tenant_id TEXT NOT NULL,
                    user_id TEXT NOT NULL,
                    surface TEXT NOT NULL,
                    route TEXT NOT NULL,
                    model TEXT NOT NULL,
                    user_message TEXT NOT NULL,
                    assistant_reply TEXT NOT NULL,
                    trace_json TEXT NOT NULL,
                    created_at TEXT NOT NULL,
                    FOREIGN KEY(thread_id) REFERENCES assistant_threads(id) ON DELETE CASCADE,
                    FOREIGN KEY(tenant_id) REFERENCES tenants(id),
                    FOREIGN KEY(user_id) REFERENCES users(id)
                );

                CREATE INDEX IF NOT EXISTS idx_assistant_runs_thread
                    ON assistant_runs(thread_id, created_at DESC);
                """
            )
            self._ensure_registry_column(connection, "users", "permissions_json", "TEXT")
            self._ensure_registry_column(connection, "tenant_fiscal_documents", "file_hash", "TEXT")
            self._ensure_registry_column(connection, "tenant_fiscal_document_inbox_items", "file_hash", "TEXT")
            self._ensure_registry_column(connection, "assistant_threads", "state_json", "TEXT")
            connection.execute(
                """
                CREATE INDEX IF NOT EXISTS idx_tenant_fiscal_documents_hash
                    ON tenant_fiscal_documents(tenant_id, file_hash)
                """
            )
            connection.execute(
                """
                CREATE INDEX IF NOT EXISTS idx_tenant_fiscal_document_inbox_items_hash
                    ON tenant_fiscal_document_inbox_items(tenant_id, file_hash)
                """
            )
            self._ensure_super_admin_account(connection)

    def _ensure_tenant_database_schema(self, connection: sqlite3.Connection) -> None:
        connection.executescript(
            """
            CREATE TABLE IF NOT EXISTS tenant_profile (
                tenant_id TEXT PRIMARY KEY,
                tenant_name TEXT NOT NULL,
                tenant_slug TEXT NOT NULL UNIQUE,
                venue_name TEXT NOT NULL,
                address TEXT NOT NULL DEFAULT '',
                phone_number TEXT,
                whatsapp_number TEXT,
                admin_name TEXT NOT NULL,
                admin_username TEXT NOT NULL,
                admin_email TEXT NOT NULL,
                admin_phone_number TEXT,
                password_hash TEXT NOT NULL,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL
            );

            CREATE TABLE IF NOT EXISTS tenant_module_settings (
                module_key TEXT PRIMARY KEY,
                enabled INTEGER NOT NULL,
                plan_name TEXT NOT NULL,
                activated_at TEXT,
                expires_at TEXT
            );

            CREATE TABLE IF NOT EXISTS tenant_timeclock_entries (
                id TEXT PRIMARY KEY,
                user_id TEXT NOT NULL,
                user_name TEXT,
                username TEXT,
                user_email TEXT,
                started_at TEXT NOT NULL,
                ended_at TEXT,
                duration_seconds INTEGER,
                started_source TEXT NOT NULL DEFAULT 'portal',
                ended_source TEXT,
                notes TEXT,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL
            );

            CREATE INDEX IF NOT EXISTS idx_tenant_timeclock_user_started
                ON tenant_timeclock_entries(user_id, started_at DESC);

            CREATE INDEX IF NOT EXISTS idx_tenant_timeclock_active
                ON tenant_timeclock_entries(ended_at, started_at DESC);

            CREATE TABLE IF NOT EXISTS tenant_inventory_warehouses (
                id TEXT PRIMARY KEY,
                name TEXT NOT NULL,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL
            );

            CREATE TABLE IF NOT EXISTS tenant_inventory_stock_items (
                id TEXT PRIMARY KEY,
                warehouse_id TEXT NOT NULL,
                product_id INTEGER,
                product_name TEXT NOT NULL,
                product_lookup TEXT NOT NULL,
                supplier_name TEXT NOT NULL,
                supplier_lookup TEXT NOT NULL,
                total_equivalent_units REAL NOT NULL DEFAULT 0,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                FOREIGN KEY (warehouse_id) REFERENCES tenant_inventory_warehouses(id) ON DELETE CASCADE,
                UNIQUE (warehouse_id, product_lookup, supplier_lookup)
            );

            CREATE TABLE IF NOT EXISTS tenant_inventory_stock_lots (
                id TEXT PRIMARY KEY,
                item_id TEXT NOT NULL,
                lot_code TEXT NOT NULL,
                lot_lookup TEXT NOT NULL,
                quantity REAL NOT NULL DEFAULT 0,
                units_per_pack REAL,
                equivalent_units REAL NOT NULL DEFAULT 0,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                FOREIGN KEY (item_id) REFERENCES tenant_inventory_stock_items(id) ON DELETE CASCADE,
                UNIQUE (item_id, lot_lookup)
            );

            CREATE INDEX IF NOT EXISTS idx_tenant_inventory_warehouse_name
                ON tenant_inventory_warehouses(name ASC);

            CREATE INDEX IF NOT EXISTS idx_tenant_inventory_items_warehouse
                ON tenant_inventory_stock_items(warehouse_id, product_name ASC);

            CREATE INDEX IF NOT EXISTS idx_tenant_inventory_lots_item
                ON tenant_inventory_stock_lots(item_id, lot_code ASC);

            CREATE TABLE IF NOT EXISTS tenant_inventory_sessions (
                id TEXT PRIMARY KEY,
                warehouse_id TEXT NOT NULL,
                warehouse_name TEXT NOT NULL,
                inventory_date TEXT NOT NULL,
                label TEXT,
                created_by_user_id TEXT,
                created_by_name TEXT,
                total_products INTEGER NOT NULL DEFAULT 0,
                total_equivalent_units REAL NOT NULL DEFAULT 0,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                FOREIGN KEY (warehouse_id) REFERENCES tenant_inventory_warehouses(id) ON DELETE CASCADE,
                UNIQUE (warehouse_id, inventory_date)
            );

            CREATE TABLE IF NOT EXISTS tenant_inventory_session_items (
                id TEXT PRIMARY KEY,
                session_id TEXT NOT NULL,
                product_id INTEGER,
                product_name TEXT NOT NULL,
                product_lookup TEXT NOT NULL,
                supplier_name TEXT NOT NULL,
                supplier_lookup TEXT NOT NULL,
                total_equivalent_units REAL NOT NULL DEFAULT 0,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                FOREIGN KEY (session_id) REFERENCES tenant_inventory_sessions(id) ON DELETE CASCADE,
                UNIQUE (session_id, product_lookup, supplier_lookup)
            );

            CREATE TABLE IF NOT EXISTS tenant_inventory_session_lots (
                id TEXT PRIMARY KEY,
                session_item_id TEXT NOT NULL,
                lot_code TEXT NOT NULL,
                lot_lookup TEXT NOT NULL,
                quantity REAL NOT NULL DEFAULT 0,
                units_per_pack REAL,
                equivalent_units REAL NOT NULL DEFAULT 0,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                FOREIGN KEY (session_item_id) REFERENCES tenant_inventory_session_items(id) ON DELETE CASCADE,
                UNIQUE (session_item_id, lot_lookup)
            );

            CREATE INDEX IF NOT EXISTS idx_tenant_inventory_session_warehouse_date
                ON tenant_inventory_sessions(warehouse_id, inventory_date DESC);

            CREATE INDEX IF NOT EXISTS idx_tenant_inventory_session_items_session
                ON tenant_inventory_session_items(session_id, product_name ASC);

            CREATE INDEX IF NOT EXISTS idx_tenant_inventory_session_lots_item
                ON tenant_inventory_session_lots(session_item_id, lot_code ASC);

            CREATE TABLE IF NOT EXISTS tenant_inventory_movements (
                id TEXT PRIMARY KEY,
                warehouse_id TEXT NOT NULL,
                product_id INTEGER,
                product_name TEXT NOT NULL,
                product_lookup TEXT NOT NULL,
                supplier_name TEXT NOT NULL,
                supplier_lookup TEXT NOT NULL,
                lot_code TEXT NOT NULL,
                lot_lookup TEXT NOT NULL,
                quantity REAL NOT NULL DEFAULT 0,
                units_per_pack REAL,
                equivalent_units REAL NOT NULL DEFAULT 0,
                movement_kind TEXT NOT NULL DEFAULT 'in',
                source_type TEXT NOT NULL DEFAULT 'manual_stock_add',
                source_label TEXT,
                occurred_at TEXT NOT NULL,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                FOREIGN KEY (warehouse_id) REFERENCES tenant_inventory_warehouses(id) ON DELETE CASCADE
            );

            CREATE INDEX IF NOT EXISTS idx_tenant_inventory_movements_warehouse_occurred
                ON tenant_inventory_movements(warehouse_id, occurred_at DESC);

            CREATE TABLE IF NOT EXISTS tenant_inventory_snapshots (
                id TEXT PRIMARY KEY,
                snapshot_date TEXT NOT NULL UNIQUE,
                label TEXT,
                created_by_user_id TEXT,
                created_by_name TEXT,
                total_warehouses INTEGER NOT NULL DEFAULT 0,
                total_products INTEGER NOT NULL DEFAULT 0,
                total_equivalent_units REAL NOT NULL DEFAULT 0,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL
            );

            CREATE TABLE IF NOT EXISTS tenant_inventory_snapshot_items (
                id TEXT PRIMARY KEY,
                snapshot_id TEXT NOT NULL,
                warehouse_id TEXT,
                warehouse_name TEXT NOT NULL,
                product_lookup TEXT NOT NULL,
                supplier_lookup TEXT NOT NULL,
                product_name TEXT NOT NULL,
                supplier_name TEXT NOT NULL,
                total_equivalent_units REAL NOT NULL DEFAULT 0,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                FOREIGN KEY (snapshot_id) REFERENCES tenant_inventory_snapshots(id) ON DELETE CASCADE,
                UNIQUE (snapshot_id, warehouse_id, product_lookup, supplier_lookup)
            );

            CREATE TABLE IF NOT EXISTS tenant_inventory_snapshot_lots (
                id TEXT PRIMARY KEY,
                snapshot_item_id TEXT NOT NULL,
                lot_code TEXT NOT NULL,
                lot_lookup TEXT NOT NULL,
                quantity REAL NOT NULL DEFAULT 0,
                units_per_pack REAL,
                equivalent_units REAL NOT NULL DEFAULT 0,
                created_at TEXT NOT NULL,
                updated_at TEXT NOT NULL,
                FOREIGN KEY (snapshot_item_id) REFERENCES tenant_inventory_snapshot_items(id) ON DELETE CASCADE,
                UNIQUE (snapshot_item_id, lot_lookup)
            );

            CREATE INDEX IF NOT EXISTS idx_tenant_inventory_snapshot_date
                ON tenant_inventory_snapshots(snapshot_date DESC);

            CREATE INDEX IF NOT EXISTS idx_tenant_inventory_snapshot_items_snapshot
                ON tenant_inventory_snapshot_items(snapshot_id, warehouse_name ASC, product_name ASC);

            CREATE INDEX IF NOT EXISTS idx_tenant_inventory_snapshot_lots_item
                ON tenant_inventory_snapshot_lots(snapshot_item_id, lot_code ASC);
            """
        )
        connection.commit()

    def _ensure_registry_column(self, connection: sqlite3.Connection, table_name: str, column_name: str, column_sql: str) -> None:
        existing_columns = {
            row["name"]
            for row in connection.execute(f"PRAGMA table_info({table_name})").fetchall()
        }
        if column_name not in existing_columns:
            connection.execute(f"ALTER TABLE {table_name} ADD COLUMN {column_name} {column_sql}")
            connection.commit()
            connection.commit()

    @contextmanager
    def _connect_registry(self):
        connection = sqlite3.connect(self._registry_path)
        connection.row_factory = sqlite3.Row
        connection.execute("PRAGMA foreign_keys = ON;")
        connection.execute("PRAGMA journal_mode = WAL;")
        try:
            yield connection
        finally:
            connection.close()

    @contextmanager
    def _connect_tenant_database(self, database_path: str):
        path = Path(database_path)
        path.parent.mkdir(parents=True, exist_ok=True)
        connection = sqlite3.connect(path)
        connection.row_factory = sqlite3.Row
        connection.execute("PRAGMA foreign_keys = ON;")
        self._ensure_tenant_database_schema(connection)
        try:
            yield connection
        finally:
            connection.close()

    def _upsert_tenant_profile(
        self,
        connection: sqlite3.Connection,
        *,
        tenant_id: str,
        tenant_name: str,
        tenant_slug: str,
        venue_name: str,
        address: str,
        phone_number: str | None,
        whatsapp_number: str | None,
        admin_name: str,
        admin_username: str,
        admin_email: str,
        admin_phone_number: str | None,
        password_hash: str,
        created_at: str,
    ) -> None:
        connection.execute(
            """
            INSERT OR REPLACE INTO tenant_profile (
                tenant_id,
                tenant_name,
                tenant_slug,
                venue_name,
                address,
                phone_number,
                whatsapp_number,
                admin_name,
                admin_username,
                admin_email,
                admin_phone_number,
                password_hash,
                created_at,
                updated_at
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                tenant_id,
                tenant_name,
                tenant_slug,
                venue_name,
                address,
                phone_number,
                whatsapp_number,
                admin_name,
                admin_username,
                admin_email,
                admin_phone_number,
                password_hash,
                created_at,
                _iso_now(),
            ),
        )

    def _ensure_super_admin_account(self, connection: sqlite3.Connection) -> None:
        created_at = _iso_now()
        database_path = str(self._tenant_dir / f"{SUPER_ADMIN_TENANT_SLUG}.sqlite3")
        tenant_exists = connection.execute(
            "SELECT id FROM tenants WHERE id = ? LIMIT 1",
            (SUPER_ADMIN_TENANT_ID,),
        ).fetchone()
        if tenant_exists is None:
            connection.execute(
                """
                INSERT INTO tenants (id, name, slug, database_path, created_at)
                VALUES (?, ?, ?, ?, ?)
                """,
                (
                    SUPER_ADMIN_TENANT_ID,
                    SUPER_ADMIN_TENANT_NAME,
                    SUPER_ADMIN_TENANT_SLUG,
                    database_path,
                    created_at,
                ),
            )

        venue_exists = connection.execute(
            "SELECT id FROM venues WHERE id = ? LIMIT 1",
            (SUPER_ADMIN_VENUE_ID,),
        ).fetchone()
        if venue_exists is None:
            connection.execute(
                """
                INSERT INTO venues (id, tenant_id, name, address, phone_number, whatsapp_number, created_at)
                VALUES (?, ?, ?, '', NULL, NULL, ?)
                """,
                (
                    SUPER_ADMIN_VENUE_ID,
                    SUPER_ADMIN_TENANT_ID,
                    SUPER_ADMIN_TENANT_NAME,
                    created_at,
                ),
            )

        user_exists = connection.execute(
            "SELECT id, password_hash, name FROM users WHERE id = ? LIMIT 1",
            (SUPER_ADMIN_USER_ID,),
        ).fetchone()
        if user_exists is None:
            connection.execute(
                """
                INSERT INTO users (id, tenant_id, name, username, email, phone_number, password_hash, role, created_at)
                VALUES (?, ?, ?, ?, ?, NULL, '', ?, ?)
                """,
                (
                    SUPER_ADMIN_USER_ID,
                    SUPER_ADMIN_TENANT_ID,
                    "Super Admin",
                    SUPER_ADMIN_USERNAME,
                    SUPER_ADMIN_EMAIL,
                    "super_admin",
                    created_at,
                ),
            )
            password_hash = ""
            admin_name = "Super Admin"
        else:
            password_hash = user_exists["password_hash"] or ""
            admin_name = user_exists["name"] or "Super Admin"
            connection.execute(
                "UPDATE users SET username = ?, email = ? WHERE id = ?",
                (SUPER_ADMIN_USERNAME, SUPER_ADMIN_EMAIL, SUPER_ADMIN_USER_ID),
            )

        self._create_tenant_database(
            database_path=database_path,
            tenant_id=SUPER_ADMIN_TENANT_ID,
            tenant_name=SUPER_ADMIN_TENANT_NAME,
            tenant_slug=SUPER_ADMIN_TENANT_SLUG,
            venue_name=SUPER_ADMIN_TENANT_NAME,
            address="",
            phone_number=None,
            whatsapp_number=None,
            admin_name=admin_name,
            admin_username=SUPER_ADMIN_USERNAME,
            admin_email=SUPER_ADMIN_EMAIL,
            admin_phone_number=None,
            password_hash=password_hash,
            created_at=created_at,
        )
        with self._connect_tenant_database(database_path) as tenant_connection:
            tenant_connection.execute("DELETE FROM tenant_module_settings")
            tenant_connection.commit()

    def _create_tenant_database(
        self,
        *,
        database_path: str,
        tenant_id: str,
        tenant_name: str,
        tenant_slug: str,
        venue_name: str,
        address: str,
        phone_number: str | None,
        whatsapp_number: str | None,
        admin_name: str,
        admin_username: str,
        admin_email: str,
        admin_phone_number: str | None,
        password_hash: str,
        created_at: str,
    ) -> None:
        with self._connect_tenant_database(database_path) as connection:
            self._upsert_tenant_profile(
                connection,
                tenant_id=tenant_id,
                tenant_name=tenant_name,
                tenant_slug=tenant_slug,
                venue_name=venue_name,
                address=address,
                phone_number=phone_number,
                whatsapp_number=whatsapp_number,
                admin_name=admin_name,
                admin_username=admin_username,
                admin_email=admin_email,
                admin_phone_number=admin_phone_number,
                password_hash=password_hash,
                created_at=created_at,
            )

            for module in MODULE_CATALOG:
                enabled = 1 if module.status == "active" else 0
                plan_name = "Platform Active" if enabled else "Planned"
                connection.execute(
                    """
                    INSERT OR REPLACE INTO tenant_module_settings (
                        module_key,
                        enabled,
                        plan_name,
                        activated_at,
                        expires_at
                    ) VALUES (?, ?, ?, ?, ?)
                    """,
                    (
                        module.key,
                        enabled,
                        plan_name,
                        created_at if enabled else None,
                        None,
                    ),
                )

            connection.commit()

    def _session_identity_from_row(self, row: sqlite3.Row) -> SessionIdentity:
        return SessionIdentity(
            token=row["token"],
            tenant_id=row["tenant_id"],
            tenant_slug=row["tenant_slug"],
            tenant_name=row["tenant_name"],
            home_tenant_id=row["home_tenant_id"],
            home_tenant_slug=row["home_tenant_slug"],
            home_tenant_name=row["home_tenant_name"],
            user_id=row["user_id"],
            user_email=row["user_email"],
            username=row["username"],
            user_name=row["user_name"],
            role=row["role"],
            permissions=_resolve_role_permissions(row["role"], row["permissions_json"] if "permissions_json" in row.keys() else None),
            database_path=row["database_path"],
            expires_at=row["expires_at"],
        )

    def _menu_asset_from_row(self, row: sqlite3.Row) -> MenuAssetRecord:
        return MenuAssetRecord(
            id=row["id"],
            tenant_id=row["tenant_id"],
            original_name=row["original_name"],
            display_name=row["display_name"],
            mime_type=row["mime_type"],
            kind=row["kind"],
            file_size_bytes=int(row["file_size_bytes"]),
            storage_path=row["storage_path"],
            extracted_text=row["extracted_text"] or "",
            analysis_text=row["analysis_text"] or "",
            status=row["status"],
            error_detail=row["error_detail"],
            created_at=row["created_at"],
            updated_at=row["updated_at"],
        )

    def _fiscal_document_from_row(self, row: sqlite3.Row) -> FiscalDocumentRecord:
        total_amount = row["total_amount"]
        return FiscalDocumentRecord(
            id=row["id"],
            tenant_id=row["tenant_id"],
            original_name=row["original_name"],
            display_name=row["display_name"],
            mime_type=row["mime_type"],
            kind=row["kind"],
            file_size_bytes=int(row["file_size_bytes"]),
            file_hash=row["file_hash"],
            storage_path=row["storage_path"],
            document_type=row["document_type"],
            document_number=row["document_number"],
            document_date=row["document_date"],
            supplier_name=row["supplier_name"],
            total_amount=float(total_amount) if total_amount is not None else None,
            currency=row["currency"] or "EUR",
            summary_text=row["summary_text"] or "",
            extracted_text=row["extracted_text"] or "",
            status=row["status"],
            matching_status=row["matching_status"] or "pending",
            review_status=row["review_status"] or "to_review",
            error_detail=row["error_detail"],
            created_at=row["created_at"],
            updated_at=row["updated_at"],
        )

    def _fiscal_document_dedup_key(self, record: FiscalDocumentRecord) -> tuple[object, ...]:
        return (
            "semantic",
            (record.document_type or "").strip().casefold(),
            (record.document_number or "").strip().casefold(),
            (record.document_date or "").strip(),
            (record.supplier_name or "").strip().casefold(),
            round(float(record.total_amount), 2) if record.total_amount is not None else None,
            (record.original_name or "").strip().casefold(),
        )

    def _assistant_thread_from_row(self, row: sqlite3.Row) -> AssistantThreadRecord:
        return AssistantThreadRecord(
            id=row["id"],
            tenant_id=row["tenant_id"],
            user_id=row["user_id"],
            surface=row["surface"],
            status=row["status"],
            state_json=row["state_json"] if "state_json" in row.keys() else None,
            created_at=row["created_at"],
            updated_at=row["updated_at"],
        )

    def _assistant_message_from_row(self, row: sqlite3.Row) -> AssistantStoredMessage:
        return AssistantStoredMessage(
            id=row["id"],
            thread_id=row["thread_id"],
            role=row["role"],
            content=row["content"],
            created_at=row["created_at"],
            sort_index=int(row["sort_index"]),
        )

    def _assistant_run_from_row(self, row: sqlite3.Row) -> AssistantRunRecord:
        return AssistantRunRecord(
            id=row["id"],
            thread_id=row["thread_id"],
            tenant_id=row["tenant_id"],
            user_id=row["user_id"],
            surface=row["surface"],
            route=row["route"],
            model=row["model"],
            user_message=row["user_message"],
            assistant_reply=row["assistant_reply"],
            trace_json=row["trace_json"],
            created_at=row["created_at"],
        )

    def _fiscal_document_settings_from_row(self, row: sqlite3.Row) -> FiscalDocumentSettingsRecord:
        return FiscalDocumentSettingsRecord(
            tenant_id=row["tenant_id"],
            inbound_email=row["inbound_email"],
            updated_at=row["updated_at"],
        )

    def _fiscal_document_inbox_item_from_row(self, row: sqlite3.Row) -> FiscalDocumentInboxItemRecord:
        return FiscalDocumentInboxItemRecord(
            id=row["id"],
            tenant_id=row["tenant_id"],
            message_id=row["message_id"],
            attachment_id=row["attachment_id"],
            file_hash=row["file_hash"],
            subject=row["subject"] or "",
            sender=row["sender"],
            received_at=row["received_at"],
            attachment_name=row["attachment_name"],
            mime_type=row["mime_type"],
            sync_status=row["sync_status"],
            document_id=row["document_id"],
            error_detail=row["error_detail"],
            created_at=row["created_at"],
            updated_at=row["updated_at"],
        )

    def _timeclock_entry_from_row(self, row: sqlite3.Row) -> TimeclockEntryRecord:
        duration_seconds = row["duration_seconds"]
        return TimeclockEntryRecord(
            id=row["id"],
            user_id=row["user_id"],
            user_name=row["user_name"],
            username=row["username"],
            user_email=row["user_email"],
            started_at=row["started_at"],
            ended_at=row["ended_at"],
            duration_seconds=int(duration_seconds) if duration_seconds is not None else None,
            started_source=row["started_source"] or "portal",
            ended_source=row["ended_source"],
            notes=row["notes"],
            created_at=row["created_at"],
            updated_at=row["updated_at"],
        )

    def _create_session(self, connection: sqlite3.Connection, *, tenant_id: str, user_id: str) -> SessionIdentity:
        token = secrets.token_urlsafe(32)
        created_at = _iso_now()
        expires_at = (_utcnow() + self._session_duration).isoformat()
        connection.execute(
            """
            INSERT INTO sessions (token, tenant_id, user_id, created_at, expires_at)
            VALUES (?, ?, ?, ?, ?)
            """,
            (token, tenant_id, user_id, created_at, expires_at),
        )
        connection.commit()

        return self.get_session(token)  # type: ignore[return-value]

    def _lookup_user_for_login(self, connection: sqlite3.Connection, identifier: str) -> sqlite3.Row | None:
        normalized = _normalize_lookup(identifier)
        return connection.execute(
            """
            SELECT
                users.*,
                tenants.name AS tenant_name,
                tenants.slug AS tenant_slug,
                tenants.database_path AS database_path
            FROM users
            JOIN tenants ON tenants.id = users.tenant_id
            WHERE lower(users.email) = ? OR lower(users.username) = ?
            LIMIT 1
            """,
            (normalized, normalized),
        ).fetchone()

    def _lookup_login_candidates(self, connection: sqlite3.Connection, identifier: str) -> list[sqlite3.Row]:
        normalized = _normalize_lookup(identifier)
        return list(
            connection.execute(
                """
                SELECT
                    users.*,
                    tenants.name AS tenant_name,
                    tenants.slug AS tenant_slug,
                    tenants.database_path AS database_path
                FROM users
                JOIN tenants ON tenants.id = users.tenant_id
                WHERE lower(users.email) = ? OR lower(users.username) = ?
                ORDER BY
                    CASE
                        WHEN lower(users.username) = ? THEN 0
                        WHEN lower(users.email) = ? THEN 1
                        ELSE 2
                    END,
                    CASE WHEN users.role = 'super_admin' THEN 0 ELSE 1 END,
                    users.created_at ASC
                """,
                (normalized, normalized, normalized, normalized),
            )
        )

    def _context_rows(self, connection: sqlite3.Connection, tenant_id: str) -> tuple[sqlite3.Row, list[sqlite3.Row], list[sqlite3.Row], list[sqlite3.Row]]:
        tenant = connection.execute("SELECT * FROM tenants WHERE id = ?", (tenant_id,)).fetchone()
        if tenant is None:
            raise KeyError(f"Tenant {tenant_id} non trovato")

        venues = list(connection.execute("SELECT * FROM venues WHERE tenant_id = ? ORDER BY created_at ASC", (tenant_id,)))
        users = list(connection.execute("SELECT * FROM users WHERE tenant_id = ? ORDER BY created_at ASC", (tenant_id,)))
        modules = list(
            connection.execute(
                "SELECT * FROM tenant_modules WHERE tenant_id = ? ORDER BY enabled DESC, module_key ASC",
                (tenant_id,),
            )
        )
        return tenant, venues, users, modules

    def _normalize_register_like_payload(
        self,
        payload: RegisterTenantPayload | AdminCreateTenantPayload | AdminUpdateTenantAdminPayload,
    ) -> dict[str, str | None]:
        return {
            "locale_name": payload.locale_name.strip(),
            "address": payload.address.strip(),
            "phone_number": payload.phone_number.strip() if payload.phone_number else None,
            "whatsapp_number": payload.whatsapp_number.strip() if payload.whatsapp_number else None,
            "admin_name": payload.admin_name.strip(),
            "username": payload.username.strip(),
            "email": payload.email.strip(),
        }

    def _normalize_tenant_staff_payload(
        self,
        payload: TenantStaffUserCreatePayload | TenantStaffUserUpdatePayload,
    ) -> dict[str, str | None]:
        return {
            "name": payload.name.strip(),
            "username": payload.username.strip(),
            "email": payload.email.strip(),
            "phone_number": payload.phone_number.strip() if payload.phone_number else None,
        }

    def _can_manage_tenant_accounts(self, session: SessionIdentity) -> bool:
        if session.tenant_id == SUPER_ADMIN_TENANT_ID:
            return False
        return session.role in {"owner", "super_admin"}

    def _can_access_timeclock(self, session: SessionIdentity) -> bool:
        if session.tenant_id == SUPER_ADMIN_TENANT_ID:
            return False
        return session.role in {"owner", "super_admin"} or self.session_has_permission(session, "timeclock")

    def _parse_iso_datetime(self, raw_value: str | None) -> datetime | None:
        if not raw_value:
            return None
        try:
            return datetime.fromisoformat(raw_value)
        except ValueError:
            return None

    def _serialize_timeclock_entry(self, record: TimeclockEntryRecord, *, reference_now: datetime | None = None) -> dict[str, object]:
        started_at = self._parse_iso_datetime(record.started_at)
        ended_at = self._parse_iso_datetime(record.ended_at)
        now = reference_now or _utcnow()
        duration_seconds = record.duration_seconds
        if duration_seconds is None and started_at is not None:
            end_reference = ended_at or now
            duration_seconds = max(int((end_reference - started_at).total_seconds()), 0)
        entry_date = started_at.date().isoformat() if started_at is not None else None
        return {
            "id": record.id,
            "user_id": record.user_id,
            "user_name": record.user_name,
            "username": record.username,
            "user_email": record.user_email,
            "started_at": record.started_at,
            "ended_at": record.ended_at,
            "duration_seconds": duration_seconds,
            "duration_hours": round((duration_seconds or 0) / 3600, 2),
            "is_active": record.ended_at is None,
            "started_source": record.started_source,
            "ended_source": record.ended_source,
            "notes": record.notes,
            "entry_date": entry_date,
            "created_at": record.created_at,
            "updated_at": record.updated_at,
        }

    def _read_timeclock_entries(
        self,
        session: SessionIdentity,
        *,
        user_id: str | None = None,
        start_date: str | None = None,
        end_date: str | None = None,
        include_active_only: bool = False,
        limit: int = 200,
    ) -> list[TimeclockEntryRecord]:
        filters: list[str] = []
        values: list[object] = []
        if user_id:
            filters.append("user_id = ?")
            values.append(user_id)
        if start_date:
            filters.append("date(started_at) >= date(?)")
            values.append(start_date)
        if end_date:
            filters.append("date(started_at) <= date(?)")
            values.append(end_date)
        if include_active_only:
            filters.append("ended_at IS NULL")
        where_sql = f"WHERE {' AND '.join(filters)}" if filters else ""

        with self._connect_tenant_database(session.database_path) as connection:
            rows = connection.execute(
                f"""
                SELECT *
                FROM tenant_timeclock_entries
                {where_sql}
                ORDER BY started_at DESC
                LIMIT ?
                """,
                (*values, max(1, min(limit, 2000))),
            ).fetchall()

        return [self._timeclock_entry_from_row(row) for row in rows]

    def _build_timeclock_summary(
        self,
        entries: list[TimeclockEntryRecord],
        *,
        now: datetime | None = None,
    ) -> list[dict[str, object]]:
        reference_now = now or _utcnow()
        grouped: dict[str, dict[str, object]] = {}
        for entry in entries:
            serialized = self._serialize_timeclock_entry(entry, reference_now=reference_now)
            bucket = grouped.setdefault(
                entry.user_id,
                {
                    "user_id": entry.user_id,
                    "user_name": entry.user_name,
                    "username": entry.username,
                    "user_email": entry.user_email,
                    "entries": 0,
                    "total_seconds": 0,
                    "active_entries": 0,
                    "last_started_at": None,
                },
            )
            bucket["entries"] = int(bucket["entries"]) + 1
            bucket["total_seconds"] = int(bucket["total_seconds"]) + int(serialized["duration_seconds"] or 0)
            if serialized["is_active"]:
                bucket["active_entries"] = int(bucket["active_entries"]) + 1
            if bucket["last_started_at"] is None or str(serialized["started_at"]) > str(bucket["last_started_at"]):
                bucket["last_started_at"] = serialized["started_at"]

        summaries: list[dict[str, object]] = []
        for item in grouped.values():
            total_seconds = int(item["total_seconds"] or 0)
            summaries.append(
                {
                    **item,
                    "total_hours": round(total_seconds / 3600, 2),
                }
            )
        summaries.sort(
            key=lambda item: (
                str(item.get("user_name") or item.get("username") or item.get("user_email") or "").lower(),
                str(item.get("user_id") or ""),
            )
        )
        return summaries

    def session_has_permission(self, session: SessionIdentity, permission: str) -> bool:
        normalized = permission.strip().lower()
        if session.role in {"owner", "super_admin"}:
            return True
        return normalized in session.permissions

    def _can_access_inventory(self, session: SessionIdentity) -> bool:
        return session.role in {"owner", "super_admin"} or self.session_has_permission(session, "inventory")

    def _tenant_table_exists(self, connection: sqlite3.Connection, table_name: str) -> bool:
        row = connection.execute(
            "SELECT name FROM sqlite_master WHERE type = 'table' AND name = ? LIMIT 1",
            (table_name,),
        ).fetchone()
        return row is not None

    def _tenant_table_columns(self, connection: sqlite3.Connection, table_name: str) -> set[str]:
        if not self._tenant_table_exists(connection, table_name):
            return set()
        return {
            str(row["name"])
            for row in connection.execute(f"PRAGMA table_info({table_name})").fetchall()
        }

    def _serialize_inventory_warehouse_row(self, row: sqlite3.Row) -> dict[str, object]:
        return {
            "id": str(row["id"]),
            "name": str(row["name"]),
            "product_count": int(row["product_count"] or 0),
            "total_equivalent_units": float(row["total_equivalent_units"] or 0),
            "created_at": str(row["created_at"]),
            "updated_at": str(row["updated_at"]),
        }

    def _serialize_inventory_stock_item_row(
        self,
        row: sqlite3.Row,
        lots: list[dict[str, object]],
    ) -> dict[str, object]:
        return {
            "id": str(row["id"]),
            "product_id": int(row["product_id"]) if row["product_id"] is not None else None,
            "product_name": str(row["product_name"]),
            "supplier_name": str(row["supplier_name"]),
            "total_equivalent_units": float(row["total_equivalent_units"] or 0),
            "lot_count": len(lots),
            "lots": lots,
            "created_at": str(row["created_at"]),
            "updated_at": str(row["updated_at"]),
        }

    def _serialize_inventory_session_row(self, row: sqlite3.Row) -> dict[str, object]:
        return {
            "id": str(row["id"]),
            "warehouse_id": str(row["warehouse_id"]),
            "warehouse_name": str(row["warehouse_name"] or ""),
            "inventory_date": str(row["inventory_date"]),
            "label": str(row["label"] or ""),
            "created_by_name": str(row["created_by_name"] or ""),
            "total_products": int(row["total_products"] or 0),
            "total_equivalent_units": float(row["total_equivalent_units"] or 0),
            "created_at": str(row["created_at"]),
            "updated_at": str(row["updated_at"]),
        }

    def _serialize_inventory_snapshot_row(self, row: sqlite3.Row) -> dict[str, object]:
        return {
            "id": str(row["id"]),
            "snapshot_date": str(row["snapshot_date"]),
            "label": str(row["label"] or ""),
            "created_by_name": str(row["created_by_name"] or ""),
            "total_warehouses": int(row["total_warehouses"] or 0),
            "total_products": int(row["total_products"] or 0),
            "total_equivalent_units": float(row["total_equivalent_units"] or 0),
            "created_at": str(row["created_at"]),
            "updated_at": str(row["updated_at"]),
        }

    def _inventory_lot_requires_units_per_pack(self, lot_code: str | None) -> bool:
        return _normalize_lookup(str(lot_code or "")) in {"ct", "cartone", "cartoni", "cassa", "casse"}

    def _inventory_effective_units(
        self,
        *,
        lot_code: str | None,
        quantity: float,
        units_per_pack: float | None,
    ) -> float:
        safe_quantity = float(quantity)
        if safe_quantity < 0:
            raise ValueError("La quantita non puo essere negativa.")
        if safe_quantity == 0:
            return 0.0
        if self._inventory_lot_requires_units_per_pack(lot_code):
            if units_per_pack is None or float(units_per_pack) <= 0:
                raise ValueError(
                    "Per questo lotto manca il dato unita per pack nel catalogo prodotti. "
                    "Completa prima il prodotto e poi aggiungilo in inventario."
                )
            return safe_quantity * float(units_per_pack)
        return safe_quantity

    def _normalize_inventory_date(self, value: str | None) -> str:
        raw_value = (value or "").strip()
        if not raw_value:
            return _utcnow().date().isoformat()
        try:
            return date.fromisoformat(raw_value).isoformat()
        except ValueError as exc:
            raise ValueError("Data inventario non valida. Usa il formato YYYY-MM-DD.") from exc

    def _normalize_inventory_snapshot_date(self, value: str | None) -> str:
        return self._normalize_inventory_date(value)

    def _parse_inventory_datetime(self, value: object | None) -> datetime | None:
        raw_value = str(value or "").strip()
        if not raw_value:
            return None

        candidates = [raw_value]
        if raw_value.endswith("Z"):
            candidates.append(f"{raw_value[:-1]}+00:00")

        for candidate in candidates:
            try:
                parsed = datetime.fromisoformat(candidate)
            except ValueError:
                continue
            if parsed.tzinfo is None:
                parsed = parsed.replace(tzinfo=timezone.utc)
            return parsed.astimezone(timezone.utc)
        return None

    def _read_inventory_warehouse_row(self, connection: sqlite3.Connection, warehouse_id: str) -> sqlite3.Row:
        row = connection.execute(
            """
            SELECT
                warehouses.*,
                COUNT(items.id) AS product_count,
                COALESCE(SUM(items.total_equivalent_units), 0) AS total_equivalent_units
            FROM tenant_inventory_warehouses AS warehouses
            LEFT JOIN tenant_inventory_stock_items AS items
                ON items.warehouse_id = warehouses.id
            WHERE warehouses.id = ?
            GROUP BY warehouses.id
            LIMIT 1
            """,
            (warehouse_id,),
        ).fetchone()
        if row is None:
            raise KeyError("Magazzino non trovato.")
        return row

    def _read_inventory_snapshot_row(self, connection: sqlite3.Connection, snapshot_id: str) -> sqlite3.Row:
        row = connection.execute(
            "SELECT * FROM tenant_inventory_snapshots WHERE id = ? LIMIT 1",
            (snapshot_id,),
        ).fetchone()
        if row is None:
            raise KeyError("Fotografia inventario non trovata.")
        return row

    def _read_inventory_session_row(self, connection: sqlite3.Connection, session_id: str) -> sqlite3.Row:
        row = connection.execute(
            "SELECT * FROM tenant_inventory_sessions WHERE id = ? LIMIT 1",
            (session_id,),
        ).fetchone()
        if row is None:
            raise KeyError("Inventario magazzino non trovato.")
        return row

    def _read_inventory_latest_session_row(
        self,
        connection: sqlite3.Connection,
        warehouse_id: str,
    ) -> sqlite3.Row | None:
        return connection.execute(
            """
            SELECT *
            FROM tenant_inventory_sessions
            WHERE warehouse_id = ?
            ORDER BY inventory_date DESC, created_at DESC
            LIMIT 1
            """,
            (warehouse_id,),
        ).fetchone()

    def _load_inventory_stock_items(
        self,
        connection: sqlite3.Connection,
        warehouse_id: str,
    ) -> tuple[list[sqlite3.Row], dict[str, list[dict[str, object]]]]:
        item_rows = connection.execute(
            """
            SELECT *
            FROM tenant_inventory_stock_items
            WHERE warehouse_id = ?
            ORDER BY lower(supplier_name) ASC, lower(product_name) ASC
            """,
            (warehouse_id,),
        ).fetchall()

        item_ids = [str(row["id"]) for row in item_rows]
        lots_by_item: dict[str, list[dict[str, object]]] = {item_id: [] for item_id in item_ids}
        if item_ids:
            placeholders = ", ".join("?" for _ in item_ids)
            lot_rows = connection.execute(
                f"""
                SELECT *
                FROM tenant_inventory_stock_lots
                WHERE item_id IN ({placeholders})
                ORDER BY lower(lot_code) ASC, created_at ASC
                """,
                tuple(item_ids),
            ).fetchall()
            for row in lot_rows:
                item_id = str(row["item_id"])
                lots_by_item.setdefault(item_id, []).append(
                    {
                        "id": str(row["id"]),
                        "lot_code": str(row["lot_code"]),
                        "quantity": float(row["quantity"] or 0),
                        "units_per_pack": float(row["units_per_pack"]) if row["units_per_pack"] is not None else None,
                        "equivalent_units": float(row["equivalent_units"] or 0),
                        "created_at": str(row["created_at"]),
                        "updated_at": str(row["updated_at"]),
                    }
                )
        return item_rows, lots_by_item

    def _load_inventory_session_items(
        self,
        connection: sqlite3.Connection,
        session_id: str,
    ) -> tuple[list[sqlite3.Row], dict[str, list[dict[str, object]]]]:
        item_rows = connection.execute(
            """
            SELECT *
            FROM tenant_inventory_session_items
            WHERE session_id = ?
            ORDER BY lower(supplier_name) ASC, lower(product_name) ASC
            """,
            (session_id,),
        ).fetchall()

        item_ids = [str(row["id"]) for row in item_rows]
        lots_by_item: dict[str, list[dict[str, object]]] = {item_id: [] for item_id in item_ids}
        if item_ids:
            placeholders = ", ".join("?" for _ in item_ids)
            lot_rows = connection.execute(
                f"""
                SELECT *
                FROM tenant_inventory_session_lots
                WHERE session_item_id IN ({placeholders})
                ORDER BY lower(lot_code) ASC, created_at ASC
                """,
                tuple(item_ids),
            ).fetchall()
            for row in lot_rows:
                item_id = str(row["session_item_id"])
                lots_by_item.setdefault(item_id, []).append(
                    {
                        "id": str(row["id"]),
                        "lot_code": str(row["lot_code"]),
                        "quantity": float(row["quantity"] or 0),
                        "units_per_pack": float(row["units_per_pack"]) if row["units_per_pack"] is not None else None,
                        "equivalent_units": float(row["equivalent_units"] or 0),
                        "created_at": str(row["created_at"]),
                        "updated_at": str(row["updated_at"]),
                    }
                )
        return item_rows, lots_by_item

    def _aggregate_inventory_snapshot_items(
        self,
        connection: sqlite3.Connection,
        snapshot_id: str,
    ) -> dict[tuple[str, str], dict[str, object]]:
        rows = connection.execute(
            """
            SELECT
                product_lookup,
                supplier_lookup,
                MIN(product_name) AS product_name,
                MIN(supplier_name) AS supplier_name,
                COALESCE(SUM(total_equivalent_units), 0) AS total_equivalent_units
            FROM tenant_inventory_snapshot_items
            WHERE snapshot_id = ?
            GROUP BY product_lookup, supplier_lookup
            """,
            (snapshot_id,),
        ).fetchall()

        aggregated: dict[tuple[str, str], dict[str, object]] = {}
        for row in rows:
            key = (str(row["product_lookup"]), str(row["supplier_lookup"]))
            aggregated[key] = {
                "product_name": str(row["product_name"] or ""),
                "supplier_name": str(row["supplier_name"] or ""),
                "total_equivalent_units": float(row["total_equivalent_units"] or 0),
            }
        return aggregated

    def _aggregate_inventory_session_items(
        self,
        connection: sqlite3.Connection,
        session_id: str,
    ) -> dict[tuple[str, str], dict[str, object]]:
        rows = connection.execute(
            """
            SELECT
                product_lookup,
                supplier_lookup,
                MIN(product_name) AS product_name,
                MIN(supplier_name) AS supplier_name,
                COALESCE(SUM(total_equivalent_units), 0) AS total_equivalent_units
            FROM tenant_inventory_session_items
            WHERE session_id = ?
            GROUP BY product_lookup, supplier_lookup
            """,
            (session_id,),
        ).fetchall()

        aggregated: dict[tuple[str, str], dict[str, object]] = {}
        for row in rows:
            key = (str(row["product_lookup"]), str(row["supplier_lookup"]))
            aggregated[key] = {
                "product_name": str(row["product_name"] or ""),
                "supplier_name": str(row["supplier_name"] or ""),
                "total_equivalent_units": float(row["total_equivalent_units"] or 0),
            }
        return aggregated

    def _sum_inventory_movements_since(
        self,
        connection: sqlite3.Connection,
        warehouse_id: str,
        *,
        start_value: object | None,
    ) -> float:
        if start_value is None:
            return 0.0
        start_dt = self._parse_inventory_datetime(start_value)
        if start_dt is None:
            raw_value = str(start_value or "").strip()
            if not raw_value:
                return 0.0
            start_dt = datetime.combine(date.fromisoformat(raw_value), time.min, tzinfo=timezone.utc)
        rows = connection.execute(
            """
            SELECT occurred_at, equivalent_units
            FROM tenant_inventory_movements
            WHERE warehouse_id = ? AND movement_kind = 'in'
            ORDER BY occurred_at ASC, created_at ASC
            """,
            (warehouse_id,),
        ).fetchall()

        total = 0.0
        for row in rows:
            occurred_at = self._parse_inventory_datetime(row["occurred_at"])
            if occurred_at is None or occurred_at < start_dt:
                continue
            total += float(row["equivalent_units"] or 0)
        return round(total, 2)

    def _build_inventory_item_comparison(
        self,
        *,
        current_items: list[dict[str, object]],
        real_items: list[dict[str, object]],
    ) -> list[dict[str, object]]:
        current_map = {
            (_normalize_lookup(str(item["product_name"])), _normalize_lookup(str(item["supplier_name"]))): item
            for item in current_items
        }
        real_map = {
            (_normalize_lookup(str(item["product_name"])), _normalize_lookup(str(item["supplier_name"]))): item
            for item in real_items
        }
        keys = set(current_map) | set(real_map)
        rows: list[dict[str, object]] = []
        for key in keys:
            current_item = current_map.get(key)
            real_item = real_map.get(key)
            product_name = str((current_item or real_item or {}).get("product_name") or "")
            supplier_name = str((current_item or real_item or {}).get("supplier_name") or "")
            theoretical_units = float((current_item or {}).get("total_equivalent_units") or 0)
            real_units = float((real_item or {}).get("total_equivalent_units") or 0)
            delta_units = theoretical_units - real_units
            if theoretical_units == 0 and real_units == 0:
                continue
            rows.append(
                {
                    "product_name": product_name,
                    "supplier_name": supplier_name,
                    "theoretical_units": round(theoretical_units, 2),
                    "real_units": round(real_units, 2),
                    "delta_units": round(delta_units, 2),
                }
            )

        rows.sort(
            key=lambda item: (
                -abs(float(item["delta_units"])),
                str(item["supplier_name"]).lower(),
                str(item["product_name"]).lower(),
            )
        )
        return rows

    def list_inventory_warehouses(self, session: SessionIdentity) -> dict[str, object]:
        if not self._can_access_inventory(session):
            raise ValueError("Il tuo account non puo accedere a Inventario.")

        with self._connect_tenant_database(session.database_path) as connection:
            rows = connection.execute(
                """
                SELECT
                    warehouses.*,
                    COUNT(items.id) AS product_count,
                    COALESCE(SUM(items.total_equivalent_units), 0) AS total_equivalent_units
                FROM tenant_inventory_warehouses AS warehouses
                LEFT JOIN tenant_inventory_stock_items AS items
                    ON items.warehouse_id = warehouses.id
                GROUP BY warehouses.id
                ORDER BY LOWER(warehouses.name) ASC, warehouses.created_at ASC
                """
            ).fetchall()

            warehouses: list[dict[str, object]] = []
            for row in rows:
                warehouse_id = str(row["id"])
                latest_session_row = self._read_inventory_latest_session_row(connection, warehouse_id)
                latest_inventory = (
                    self._serialize_inventory_session_row(latest_session_row)
                    if latest_session_row is not None
                    else None
                )
                session_count_row = connection.execute(
                    "SELECT COUNT(*) AS total FROM tenant_inventory_sessions WHERE warehouse_id = ?",
                    (warehouse_id,),
                ).fetchone()
                warehouse_summary = self._serialize_inventory_warehouse_row(row)
                warehouse_summary["theoretical_total_equivalent_units"] = float(row["total_equivalent_units"] or 0)
                warehouse_summary["real_total_equivalent_units"] = (
                    float(latest_session_row["total_equivalent_units"] or 0)
                    if latest_session_row is not None
                    else 0.0
                )
                warehouse_summary["latest_inventory_date"] = (
                    str(latest_session_row["inventory_date"])
                    if latest_session_row is not None
                    else None
                )
                warehouse_summary["latest_inventory"] = latest_inventory
                warehouse_summary["inventory_session_count"] = int(session_count_row["total"] or 0) if session_count_row is not None else 0
                warehouse_summary["incoming_since_latest_inventory_equivalent_units"] = self._sum_inventory_movements_since(
                    connection,
                warehouse_id,
                    start_value=latest_session_row["created_at"] if latest_session_row is not None else None,
                )
                warehouses.append(warehouse_summary)

        return {
            "warehouses": warehouses,
            "warehouse_count": len(warehouses),
            "total_products": sum(int(item["product_count"]) for item in warehouses),
            "total_equivalent_units": sum(float(item["total_equivalent_units"]) for item in warehouses),
            "theoretical_total_equivalent_units": sum(float(item["theoretical_total_equivalent_units"]) for item in warehouses),
            "real_total_equivalent_units": sum(float(item["real_total_equivalent_units"]) for item in warehouses),
            "inventory_session_count": sum(int(item["inventory_session_count"]) for item in warehouses),
        }

    def create_inventory_warehouse(
        self,
        session: SessionIdentity,
        payload: InventoryWarehouseCreatePayload,
    ) -> dict[str, object]:
        if not self._can_access_inventory(session):
            raise ValueError("Il tuo account non puo accedere a Inventario.")

        name = payload.name.strip()
        if len(name) < 2:
            raise ValueError("Dammi un nome magazzino valido.")

        warehouse_id = f"inventory_wh_{uuid.uuid4().hex}"
        timestamp = _iso_now()
        with self._lock:
            with self._connect_tenant_database(session.database_path) as connection:
                existing_row = connection.execute(
                    "SELECT id FROM tenant_inventory_warehouses WHERE lower(name) = lower(?) LIMIT 1",
                    (name,),
                ).fetchone()
                if existing_row is not None:
                    raise ValueError("Esiste gia un magazzino con questo nome.")
                connection.execute(
                    """
                    INSERT INTO tenant_inventory_warehouses (id, name, created_at, updated_at)
                    VALUES (?, ?, ?, ?)
                    """,
                    (warehouse_id, name, timestamp, timestamp),
                )
                connection.commit()
                row = self._read_inventory_warehouse_row(connection, warehouse_id)

        return {"warehouse": self._serialize_inventory_warehouse_row(row)}

    def list_inventory_warehouse_sessions(
        self,
        session: SessionIdentity,
        warehouse_id: str,
    ) -> dict[str, object]:
        if not self._can_access_inventory(session):
            raise ValueError("Il tuo account non puo accedere a Inventario.")

        with self._connect_tenant_database(session.database_path) as connection:
            warehouse_row = self._read_inventory_warehouse_row(connection, warehouse_id)
            rows = connection.execute(
                """
                SELECT *
                FROM tenant_inventory_sessions
                WHERE warehouse_id = ?
                ORDER BY inventory_date DESC, created_at DESC
                """,
                (warehouse_id,),
            ).fetchall()

        sessions = [self._serialize_inventory_session_row(row) for row in rows]
        return {
            "warehouse": self._serialize_inventory_warehouse_row(warehouse_row),
            "sessions": sessions,
            "total_count": len(sessions),
        }

    def get_inventory_warehouse_session_detail(
        self,
        session: SessionIdentity,
        warehouse_id: str,
        session_id: str,
    ) -> dict[str, object]:
        if not self._can_access_inventory(session):
            raise ValueError("Il tuo account non puo accedere a Inventario.")

        with self._connect_tenant_database(session.database_path) as connection:
            warehouse_row = self._read_inventory_warehouse_row(connection, warehouse_id)
            session_row = self._read_inventory_session_row(connection, session_id)
            if str(session_row["warehouse_id"] or "") != warehouse_id:
                raise ValueError("L'inventario richiesto non appartiene a questo magazzino.")
            item_rows, lots_by_item = self._load_inventory_session_items(connection, session_id)

        items = [
            self._serialize_inventory_stock_item_row(row, lots_by_item.get(str(row["id"]), []))
            for row in item_rows
        ]
        return {
            "warehouse": self._serialize_inventory_warehouse_row(warehouse_row),
            "session": self._serialize_inventory_session_row(session_row),
            "items": items,
            "total_products": len(items),
            "total_equivalent_units": sum(float(item["total_equivalent_units"]) for item in items),
        }

    def create_inventory_warehouse_session(
        self,
        session: SessionIdentity,
        warehouse_id: str,
        payload: InventorySessionCreatePayload,
    ) -> dict[str, object]:
        if not self._can_access_inventory(session):
            raise ValueError("Il tuo account non puo accedere a Inventario.")

        inventory_date = self._normalize_inventory_date(payload.inventory_date)
        timestamp = _iso_now()
        actor_name = (session.user_name or session.username or session.user_email or "Utente locale").strip()
        session_id = f"inventory_session_{uuid.uuid4().hex}"
        replaced_existing = False

        with self._lock:
            with self._connect_tenant_database(session.database_path) as connection:
                warehouse_row = self._read_inventory_warehouse_row(connection, warehouse_id)
                existing_row = connection.execute(
                    """
                    SELECT id
                    FROM tenant_inventory_sessions
                    WHERE warehouse_id = ? AND inventory_date = ?
                    LIMIT 1
                    """,
                    (warehouse_id, inventory_date),
                ).fetchone()
                if existing_row is not None:
                    replaced_existing = True
                    connection.execute(
                        "DELETE FROM tenant_inventory_sessions WHERE id = ?",
                        (str(existing_row["id"]),),
                    )

                item_rows, lots_by_item = self._load_inventory_stock_items(connection, warehouse_id)
                total_equivalent_units = sum(float(row["total_equivalent_units"] or 0) for row in item_rows)
                connection.execute(
                    """
                    INSERT INTO tenant_inventory_sessions (
                        id,
                        warehouse_id,
                        warehouse_name,
                        inventory_date,
                        label,
                        created_by_user_id,
                        created_by_name,
                        total_products,
                        total_equivalent_units,
                        created_at,
                        updated_at
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    (
                        session_id,
                        warehouse_id,
                        str(warehouse_row["name"] or ""),
                        inventory_date,
                        f"Inventario {warehouse_row['name']} del {inventory_date}",
                        session.user_id,
                        actor_name,
                        len(item_rows),
                        total_equivalent_units,
                        timestamp,
                        timestamp,
                    ),
                )

                for item_row in item_rows:
                    session_item_id = f"inventory_session_item_{uuid.uuid4().hex}"
                    connection.execute(
                        """
                        INSERT INTO tenant_inventory_session_items (
                            id,
                            session_id,
                            product_id,
                            product_name,
                            product_lookup,
                            supplier_name,
                            supplier_lookup,
                            total_equivalent_units,
                            created_at,
                            updated_at
                        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                        """,
                        (
                            session_item_id,
                            session_id,
                            int(item_row["product_id"]) if item_row["product_id"] is not None else None,
                            str(item_row["product_name"] or ""),
                            str(item_row["product_lookup"] or ""),
                            str(item_row["supplier_name"] or ""),
                            str(item_row["supplier_lookup"] or ""),
                            float(item_row["total_equivalent_units"] or 0),
                            timestamp,
                            timestamp,
                        ),
                    )
                    for lot_row in lots_by_item.get(str(item_row["id"]), []):
                        connection.execute(
                            """
                            INSERT INTO tenant_inventory_session_lots (
                                id,
                                session_item_id,
                                lot_code,
                                lot_lookup,
                                quantity,
                                units_per_pack,
                                equivalent_units,
                                created_at,
                                updated_at
                            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                            """,
                            (
                                f"inventory_session_lot_{uuid.uuid4().hex}",
                                session_item_id,
                                str(lot_row["lot_code"] or ""),
                                _normalize_lookup(str(lot_row["lot_code"] or "")),
                                float(lot_row["quantity"] or 0),
                                float(lot_row["units_per_pack"]) if lot_row["units_per_pack"] is not None else None,
                                float(lot_row["equivalent_units"] or 0),
                                timestamp,
                                timestamp,
                            ),
                        )

                connection.commit()
                saved_session_row = self._read_inventory_session_row(connection, session_id)

        return {
            "session": self._serialize_inventory_session_row(saved_session_row),
            "replaced_existing": replaced_existing,
        }

    def get_inventory_warehouse_consumption(
        self,
        session: SessionIdentity,
        warehouse_id: str,
        *,
        start_session_id: str | None = None,
        end_session_id: str | None = None,
    ) -> dict[str, object]:
        if not self._can_access_inventory(session):
            raise ValueError("Il tuo account non puo accedere a Inventario.")

        with self._connect_tenant_database(session.database_path) as connection:
            warehouse_row = self._read_inventory_warehouse_row(connection, warehouse_id)
            session_rows = connection.execute(
                """
                SELECT *
                FROM tenant_inventory_sessions
                WHERE warehouse_id = ?
                ORDER BY inventory_date DESC, created_at DESC
                """,
                (warehouse_id,),
            ).fetchall()
            serialized_sessions = [self._serialize_inventory_session_row(row) for row in session_rows]
            if not session_rows:
                return {
                    "warehouse": self._serialize_inventory_warehouse_row(warehouse_row),
                    "sessions": serialized_sessions,
                    "start_session": None,
                    "end_session": None,
                    "period_days": 0,
                    "total_opening_units": 0,
                    "total_incoming_units": 0,
                    "total_closing_units": 0,
                    "total_consumed_units": 0,
                    "items": [],
                    "calculation_mode": "consumo = inventario iniziale + entrate registrate - inventario finale",
                }

            if start_session_id and end_session_id and start_session_id == end_session_id:
                raise ValueError("Per confrontare i consumi servono due inventari diversi dello stesso magazzino.")

            if start_session_id:
                start_session_row = self._read_inventory_session_row(connection, start_session_id)
            elif len(session_rows) >= 2:
                start_session_row = session_rows[1]
            else:
                start_session_row = None

            if end_session_id:
                end_session_row = self._read_inventory_session_row(connection, end_session_id)
            else:
                end_session_row = session_rows[0]

            if start_session_row is not None and str(start_session_row["warehouse_id"] or "") != warehouse_id:
                raise ValueError("L'inventario iniziale selezionato non appartiene a questo magazzino.")
            if end_session_row is not None and str(end_session_row["warehouse_id"] or "") != warehouse_id:
                raise ValueError("L'inventario finale selezionato non appartiene a questo magazzino.")

            if start_session_row is None or end_session_row is None:
                return {
                    "warehouse": self._serialize_inventory_warehouse_row(warehouse_row),
                    "sessions": serialized_sessions,
                    "start_session": self._serialize_inventory_session_row(start_session_row) if start_session_row is not None else None,
                    "end_session": self._serialize_inventory_session_row(end_session_row) if end_session_row is not None else None,
                    "period_days": 0,
                    "total_opening_units": 0,
                    "total_incoming_units": 0,
                    "total_closing_units": float(end_session_row["total_equivalent_units"] or 0) if end_session_row is not None else 0,
                    "total_consumed_units": 0,
                    "items": [],
                    "calculation_mode": "consumo = inventario iniziale + entrate registrate - inventario finale",
                }

            try:
                start_inventory_date = date.fromisoformat(str(start_session_row["inventory_date"]))
                end_inventory_date = date.fromisoformat(str(end_session_row["inventory_date"]))
            except ValueError as exc:
                raise ValueError("Non riesco a leggere correttamente le date degli inventari magazzino.") from exc

            if start_inventory_date > end_inventory_date:
                start_session_row, end_session_row = end_session_row, start_session_row
                start_inventory_date, end_inventory_date = end_inventory_date, start_inventory_date

            opening_map = self._aggregate_inventory_session_items(connection, str(start_session_row["id"]))
            closing_map = self._aggregate_inventory_session_items(connection, str(end_session_row["id"]))

            period_start_dt = self._parse_inventory_datetime(start_session_row["created_at"]) or datetime.combine(
                start_inventory_date,
                time.min,
                tzinfo=timezone.utc,
            )
            period_end_dt = self._parse_inventory_datetime(end_session_row["created_at"]) or datetime.combine(
                end_inventory_date + timedelta(days=1),
                time.min,
                tzinfo=timezone.utc,
            )
            movement_rows = connection.execute(
                """
                SELECT *
                FROM tenant_inventory_movements
                WHERE warehouse_id = ? AND movement_kind = 'in'
                ORDER BY occurred_at ASC, created_at ASC
                """,
                (warehouse_id,),
            ).fetchall()

            incoming_map: dict[tuple[str, str], dict[str, object]] = {}
            for row in movement_rows:
                occurred_at = self._parse_inventory_datetime(row["occurred_at"])
                if occurred_at is None or occurred_at <= period_start_dt or occurred_at > period_end_dt:
                    continue
                key = (str(row["product_lookup"] or ""), str(row["supplier_lookup"] or ""))
                bucket = incoming_map.setdefault(
                    key,
                    {
                        "product_name": str(row["product_name"] or ""),
                        "supplier_name": str(row["supplier_name"] or ""),
                        "total_equivalent_units": 0.0,
                    },
                )
                bucket["total_equivalent_units"] = float(bucket["total_equivalent_units"] or 0) + float(row["equivalent_units"] or 0)

        item_keys = set(opening_map) | set(closing_map) | set(incoming_map)
        items: list[dict[str, object]] = []
        for key in item_keys:
            opening = opening_map.get(key)
            closing = closing_map.get(key)
            incoming = incoming_map.get(key)
            product_name = str((closing or opening or incoming or {}).get("product_name") or "")
            supplier_name = str((closing or opening or incoming or {}).get("supplier_name") or "")
            opening_units = float((opening or {}).get("total_equivalent_units") or 0)
            incoming_units = float((incoming or {}).get("total_equivalent_units") or 0)
            closing_units = float((closing or {}).get("total_equivalent_units") or 0)
            consumed_units = opening_units + incoming_units - closing_units
            if opening_units == 0 and incoming_units == 0 and closing_units == 0:
                continue
            items.append(
                {
                    "product_name": product_name,
                    "supplier_name": supplier_name,
                    "opening_units": round(opening_units, 2),
                    "incoming_units": round(incoming_units, 2),
                    "closing_units": round(closing_units, 2),
                    "consumed_units": round(consumed_units, 2),
                }
            )

        items.sort(
            key=lambda item: (
                -float(item["consumed_units"]),
                str(item["supplier_name"]).lower(),
                str(item["product_name"]).lower(),
            )
        )
        return {
            "warehouse": self._serialize_inventory_warehouse_row(warehouse_row),
            "sessions": serialized_sessions,
            "start_session": self._serialize_inventory_session_row(start_session_row),
            "end_session": self._serialize_inventory_session_row(end_session_row),
            "period_days": max(float((end_inventory_date - start_inventory_date).days), 0.0),
            "total_opening_units": round(sum(float(item["opening_units"]) for item in items), 2),
            "total_incoming_units": round(sum(float(item["incoming_units"]) for item in items), 2),
            "total_closing_units": round(sum(float(item["closing_units"]) for item in items), 2),
            "total_consumed_units": round(sum(float(item["consumed_units"]) for item in items), 2),
            "items": items,
            "calculation_mode": "consumo = inventario iniziale + entrate registrate - inventario finale",
        }

    def search_inventory_products(
        self,
        session: SessionIdentity,
        *,
        query: str | None,
        limit: int = 20,
    ) -> dict[str, object]:
        if not self._can_access_inventory(session):
            raise ValueError("Il tuo account non puo accedere a Inventario.")

        safe_limit = max(1, min(int(limit), 100))
        normalized_query = str(query or "").strip().lower()
        with self._connect_tenant_database(session.database_path) as connection:
            if not self._tenant_table_exists(connection, "ordini_products"):
                return {"catalog_available": False, "items": []}

            product_columns = self._tenant_table_columns(connection, "ordini_products")
            select_units_per_pack = "units_per_pack" if "units_per_pack" in product_columns else "NULL AS units_per_pack"
            select_product_code = "product_code" if "product_code" in product_columns else "NULL AS product_code"
            select_price = "final_price_vat" if "final_price_vat" in product_columns else "NULL AS final_price_vat"

            if normalized_query:
                like_value = f"%{normalized_query}%"
                prefix_value = f"{normalized_query}%"
                rows = connection.execute(
                    f"""
                    SELECT
                        id,
                        product_name,
                        lot_code,
                        supplier_name,
                        {select_product_code},
                        {select_units_per_pack},
                        {select_price}
                    FROM ordini_products
                    WHERE active = 1
                      AND (
                        lower(product_name) LIKE ?
                        OR lower(supplier_name) LIKE ?
                        OR lower(lot_code) LIKE ?
                        OR lower(COALESCE(product_code, '')) LIKE ?
                      )
                    ORDER BY
                        CASE
                            WHEN lower(product_name) = ? THEN 0
                            WHEN lower(product_name) LIKE ? THEN 1
                            WHEN lower(supplier_name) LIKE ? THEN 2
                            ELSE 3
                        END,
                        lower(supplier_name) ASC,
                        lower(product_name) ASC,
                        lower(lot_code) ASC
                    LIMIT ?
                    """,
                    (
                        like_value,
                        like_value,
                        like_value,
                        like_value,
                        normalized_query,
                        prefix_value,
                        prefix_value,
                        safe_limit,
                    ),
                ).fetchall()
            else:
                rows = connection.execute(
                    f"""
                    SELECT
                        id,
                        product_name,
                        lot_code,
                        supplier_name,
                        {select_product_code},
                        {select_units_per_pack},
                        {select_price}
                    FROM ordini_products
                    WHERE active = 1
                    ORDER BY lower(supplier_name) ASC, lower(product_name) ASC, lower(lot_code) ASC
                    LIMIT ?
                    """,
                    (safe_limit,),
                ).fetchall()

        items = [
            {
                "id": int(row["id"]),
                "product_name": str(row["product_name"] or ""),
                "lot_code": str(row["lot_code"] or ""),
                "supplier_name": str(row["supplier_name"] or ""),
                "product_code": str(row["product_code"] or "") or None,
                "units_per_pack": float(row["units_per_pack"]) if row["units_per_pack"] is not None else None,
                "final_price_vat": float(row["final_price_vat"]) if row["final_price_vat"] is not None else None,
                "requires_units_per_pack": self._inventory_lot_requires_units_per_pack(row["lot_code"]),
            }
            for row in rows
        ]
        return {"catalog_available": True, "items": items}

    def list_inventory_snapshots(self, session: SessionIdentity) -> dict[str, object]:
        if not self._can_access_inventory(session):
            raise ValueError("Il tuo account non puo accedere a Inventario.")

        with self._connect_tenant_database(session.database_path) as connection:
            rows = connection.execute(
                """
                SELECT *
                FROM tenant_inventory_snapshots
                ORDER BY snapshot_date DESC, created_at DESC
                """
            ).fetchall()

        snapshots = [self._serialize_inventory_snapshot_row(row) for row in rows]
        return {"snapshots": snapshots, "total_count": len(snapshots)}

    def create_inventory_snapshot(
        self,
        session: SessionIdentity,
        payload: InventorySnapshotCreatePayload,
    ) -> dict[str, object]:
        if not self._can_access_inventory(session):
            raise ValueError("Il tuo account non puo accedere a Inventario.")

        snapshot_date = self._normalize_inventory_snapshot_date(payload.snapshot_date)
        timestamp = _iso_now()
        actor_name = (session.user_name or session.username or session.user_email or "Utente locale").strip()
        snapshot_id = f"inventory_snapshot_{uuid.uuid4().hex}"
        replaced_existing = False

        with self._lock:
            with self._connect_tenant_database(session.database_path) as connection:
                existing_row = connection.execute(
                    "SELECT id FROM tenant_inventory_snapshots WHERE snapshot_date = ? LIMIT 1",
                    (snapshot_date,),
                ).fetchone()
                if existing_row is not None:
                    replaced_existing = True
                    connection.execute(
                        "DELETE FROM tenant_inventory_snapshots WHERE id = ?",
                        (str(existing_row["id"]),),
                    )

                warehouse_rows = connection.execute(
                    """
                    SELECT *
                    FROM tenant_inventory_warehouses
                    ORDER BY lower(name) ASC, created_at ASC
                    """
                ).fetchall()
                item_rows = connection.execute(
                    """
                    SELECT
                        items.*,
                        warehouses.name AS warehouse_name
                    FROM tenant_inventory_stock_items AS items
                    JOIN tenant_inventory_warehouses AS warehouses
                        ON warehouses.id = items.warehouse_id
                    ORDER BY lower(warehouses.name) ASC, lower(items.supplier_name) ASC, lower(items.product_name) ASC
                    """
                ).fetchall()

                item_ids = [str(row["id"]) for row in item_rows]
                lots_by_item: dict[str, list[sqlite3.Row]] = {item_id: [] for item_id in item_ids}
                if item_ids:
                    placeholders = ", ".join("?" for _ in item_ids)
                    lot_rows = connection.execute(
                        f"""
                        SELECT *
                        FROM tenant_inventory_stock_lots
                        WHERE item_id IN ({placeholders})
                        ORDER BY lower(lot_code) ASC, created_at ASC
                        """,
                        tuple(item_ids),
                    ).fetchall()
                    for row in lot_rows:
                        lots_by_item.setdefault(str(row["item_id"]), []).append(row)

                total_equivalent_units = sum(float(row["total_equivalent_units"] or 0) for row in item_rows)
                connection.execute(
                    """
                    INSERT INTO tenant_inventory_snapshots (
                        id,
                        snapshot_date,
                        label,
                        created_by_user_id,
                        created_by_name,
                        total_warehouses,
                        total_products,
                        total_equivalent_units,
                        created_at,
                        updated_at
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    (
                        snapshot_id,
                        snapshot_date,
                        f"Inventario del {snapshot_date}",
                        session.user_id,
                        actor_name,
                        len(warehouse_rows),
                        len(item_rows),
                        total_equivalent_units,
                        timestamp,
                        timestamp,
                    ),
                )

                for item_row in item_rows:
                    snapshot_item_id = f"inventory_snapshot_item_{uuid.uuid4().hex}"
                    connection.execute(
                        """
                        INSERT INTO tenant_inventory_snapshot_items (
                            id,
                            snapshot_id,
                            warehouse_id,
                            warehouse_name,
                            product_lookup,
                            supplier_lookup,
                            product_name,
                            supplier_name,
                            total_equivalent_units,
                            created_at,
                            updated_at
                        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                        """,
                        (
                            snapshot_item_id,
                            snapshot_id,
                            str(item_row["warehouse_id"] or ""),
                            str(item_row["warehouse_name"] or ""),
                            str(item_row["product_lookup"] or ""),
                            str(item_row["supplier_lookup"] or ""),
                            str(item_row["product_name"] or ""),
                            str(item_row["supplier_name"] or ""),
                            float(item_row["total_equivalent_units"] or 0),
                            timestamp,
                            timestamp,
                        ),
                    )
                    for lot_row in lots_by_item.get(str(item_row["id"]), []):
                        connection.execute(
                            """
                            INSERT INTO tenant_inventory_snapshot_lots (
                                id,
                                snapshot_item_id,
                                lot_code,
                                lot_lookup,
                                quantity,
                                units_per_pack,
                                equivalent_units,
                                created_at,
                                updated_at
                            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                            """,
                            (
                                f"inventory_snapshot_lot_{uuid.uuid4().hex}",
                                snapshot_item_id,
                                str(lot_row["lot_code"] or ""),
                                str(lot_row["lot_lookup"] or ""),
                                float(lot_row["quantity"] or 0),
                                float(lot_row["units_per_pack"]) if lot_row["units_per_pack"] is not None else None,
                                float(lot_row["equivalent_units"] or 0),
                                timestamp,
                                timestamp,
                            ),
                        )

                connection.commit()
                snapshot_row = self._read_inventory_snapshot_row(connection, snapshot_id)

        return {
            "snapshot": self._serialize_inventory_snapshot_row(snapshot_row),
            "replaced_existing": replaced_existing,
        }

    def get_inventory_consumption_overview(
        self,
        session: SessionIdentity,
        *,
        start_snapshot_id: str | None = None,
        end_snapshot_id: str | None = None,
    ) -> dict[str, object]:
        if not self._can_access_inventory(session):
            raise ValueError("Il tuo account non puo accedere a Inventario.")

        with self._connect_tenant_database(session.database_path) as connection:
            snapshot_rows = connection.execute(
                """
                SELECT *
                FROM tenant_inventory_snapshots
                ORDER BY snapshot_date DESC, created_at DESC
                """
            ).fetchall()
            serialized_snapshots = [self._serialize_inventory_snapshot_row(row) for row in snapshot_rows]
            if not snapshot_rows:
                return {
                    "snapshots": serialized_snapshots,
                    "start_snapshot": None,
                    "end_snapshot": None,
                    "period_days": 0,
                    "total_opening_units": 0,
                    "total_incoming_units": 0,
                    "total_closing_units": 0,
                    "total_consumed_units": 0,
                    "incoming_data_gaps": [],
                    "items": [],
                    "calculation_mode": "consumo = inventario iniziale + ordini confermati - inventario finale",
                }

            if start_snapshot_id and end_snapshot_id and start_snapshot_id == end_snapshot_id:
                raise ValueError("Per confrontare i consumi servono due fotografie diverse.")

            if start_snapshot_id:
                start_snapshot_row = self._read_inventory_snapshot_row(connection, start_snapshot_id)
            elif len(snapshot_rows) >= 2:
                start_snapshot_row = snapshot_rows[1]
            else:
                start_snapshot_row = None

            if end_snapshot_id:
                end_snapshot_row = self._read_inventory_snapshot_row(connection, end_snapshot_id)
            else:
                end_snapshot_row = snapshot_rows[0]

            if start_snapshot_row is None or end_snapshot_row is None:
                return {
                    "snapshots": serialized_snapshots,
                    "start_snapshot": self._serialize_inventory_snapshot_row(start_snapshot_row) if start_snapshot_row is not None else None,
                    "end_snapshot": self._serialize_inventory_snapshot_row(end_snapshot_row) if end_snapshot_row is not None else None,
                    "period_days": 0,
                    "total_opening_units": 0,
                    "total_incoming_units": 0,
                    "total_closing_units": float(end_snapshot_row["total_equivalent_units"] or 0) if end_snapshot_row is not None else 0,
                    "total_consumed_units": 0,
                    "incoming_data_gaps": [],
                    "items": [],
                    "calculation_mode": "consumo = inventario iniziale + ordini confermati - inventario finale",
                }

            try:
                start_snapshot_date = date.fromisoformat(str(start_snapshot_row["snapshot_date"]))
                end_snapshot_date = date.fromisoformat(str(end_snapshot_row["snapshot_date"]))
            except ValueError as exc:
                raise ValueError("Non riesco a leggere correttamente le date delle fotografie inventario.") from exc

            if start_snapshot_date > end_snapshot_date:
                start_snapshot_row, end_snapshot_row = end_snapshot_row, start_snapshot_row
                start_snapshot_date, end_snapshot_date = end_snapshot_date, start_snapshot_date

            period_start_dt = datetime.combine(start_snapshot_date, time.min, tzinfo=timezone.utc)
            period_end_dt = datetime.combine(end_snapshot_date + timedelta(days=1), time.min, tzinfo=timezone.utc)

            opening_map = self._aggregate_inventory_snapshot_items(connection, str(start_snapshot_row["id"]))
            closing_map = self._aggregate_inventory_snapshot_items(connection, str(end_snapshot_row["id"]))

            incoming_map: dict[tuple[str, str], dict[str, object]] = {}
            incoming_data_gaps: list[dict[str, str]] = []
            has_orders_schema = self._tenant_table_exists(connection, "ordini_items") and self._tenant_table_exists(connection, "ordini_batches")
            if has_orders_schema:
                product_columns = self._tenant_table_columns(connection, "ordini_products")
                select_product_units_per_pack = "products.units_per_pack" if "units_per_pack" in product_columns else "NULL"
                broad_start_date = start_snapshot_date.isoformat()
                broad_end_date = (end_snapshot_date + timedelta(days=1)).isoformat()
                order_rows = connection.execute(
                    f"""
                    SELECT
                        items.product_name,
                        items.supplier_name,
                        items.lot_code,
                        items.quantity,
                        COALESCE(items.units_per_pack, {select_product_units_per_pack}) AS resolved_units_per_pack,
                        batches.confirmed_at
                    FROM ordini_items AS items
                    JOIN ordini_batches AS batches
                        ON batches.id = items.batch_id
                    LEFT JOIN ordini_products AS products
                        ON products.id = items.product_id
                    WHERE batches.confirmed_at >= ? AND batches.confirmed_at < ?
                    ORDER BY batches.confirmed_at ASC, items.id ASC
                    """,
                    (broad_start_date, broad_end_date),
                ).fetchall()

                seen_gap_keys: set[tuple[str, str, str]] = set()
                for row in order_rows:
                    confirmed_at = self._parse_inventory_datetime(row["confirmed_at"])
                    if confirmed_at is None or confirmed_at < period_start_dt or confirmed_at >= period_end_dt:
                        continue
                    product_name = str(row["product_name"] or "").strip()
                    supplier_name = str(row["supplier_name"] or "").strip()
                    lot_code = str(row["lot_code"] or "").strip()
                    quantity = float(row["quantity"] or 0)
                    if quantity <= 0:
                        continue
                    resolved_units_per_pack = (
                        float(row["resolved_units_per_pack"]) if row["resolved_units_per_pack"] is not None else None
                    )
                    try:
                        incoming_units = self._inventory_effective_units(
                            lot_code=lot_code,
                            quantity=quantity,
                            units_per_pack=resolved_units_per_pack,
                        )
                    except ValueError:
                        incoming_units = quantity
                        gap_key = (product_name, supplier_name, lot_code)
                        if gap_key not in seen_gap_keys:
                            seen_gap_keys.add(gap_key)
                            incoming_data_gaps.append(
                                {
                                    "product_name": product_name,
                                    "supplier_name": supplier_name,
                                    "lot_code": lot_code,
                                }
                            )

                    key = (_normalize_lookup(product_name), _normalize_lookup(supplier_name))
                    bucket = incoming_map.setdefault(
                        key,
                        {
                            "product_name": product_name,
                            "supplier_name": supplier_name,
                            "total_equivalent_units": 0.0,
                        },
                    )
                    bucket["total_equivalent_units"] = float(bucket["total_equivalent_units"] or 0) + incoming_units

            item_keys = set(opening_map) | set(closing_map) | set(incoming_map)
            items: list[dict[str, object]] = []
            for key in item_keys:
                opening = opening_map.get(key)
                closing = closing_map.get(key)
                incoming = incoming_map.get(key)
                product_name = str(
                    (closing or opening or incoming or {}).get("product_name") or ""
                )
                supplier_name = str(
                    (closing or opening or incoming or {}).get("supplier_name") or ""
                )
                opening_units = float((opening or {}).get("total_equivalent_units") or 0)
                incoming_units = float((incoming or {}).get("total_equivalent_units") or 0)
                closing_units = float((closing or {}).get("total_equivalent_units") or 0)
                consumed_units = opening_units + incoming_units - closing_units
                if opening_units == 0 and incoming_units == 0 and closing_units == 0:
                    continue
                items.append(
                    {
                        "product_name": product_name,
                        "supplier_name": supplier_name,
                        "opening_units": round(opening_units, 2),
                        "incoming_units": round(incoming_units, 2),
                        "closing_units": round(closing_units, 2),
                        "consumed_units": round(consumed_units, 2),
                    }
                )

        items.sort(
            key=lambda item: (
                -float(item["consumed_units"]),
                str(item["supplier_name"]).lower(),
                str(item["product_name"]).lower(),
            )
        )
        total_opening_units = round(sum(float(item["opening_units"]) for item in items), 2)
        total_incoming_units = round(sum(float(item["incoming_units"]) for item in items), 2)
        total_closing_units = round(sum(float(item["closing_units"]) for item in items), 2)
        total_consumed_units = round(sum(float(item["consumed_units"]) for item in items), 2)
        period_days = max(float((end_snapshot_date - start_snapshot_date).days), 0.0)

        return {
            "snapshots": serialized_snapshots,
            "start_snapshot": self._serialize_inventory_snapshot_row(start_snapshot_row),
            "end_snapshot": self._serialize_inventory_snapshot_row(end_snapshot_row),
            "period_days": period_days,
            "total_opening_units": total_opening_units,
            "total_incoming_units": total_incoming_units,
            "total_closing_units": total_closing_units,
            "total_consumed_units": total_consumed_units,
            "incoming_data_gaps": incoming_data_gaps,
            "items": items,
            "calculation_mode": "consumo = inventario iniziale + ordini confermati - inventario finale",
        }

    def get_inventory_warehouse_detail(self, session: SessionIdentity, warehouse_id: str) -> dict[str, object]:
        if not self._can_access_inventory(session):
            raise ValueError("Il tuo account non puo accedere a Inventario.")

        with self._connect_tenant_database(session.database_path) as connection:
            warehouse_row = self._read_inventory_warehouse_row(connection, warehouse_id)
            current_item_rows, current_lots_by_item = self._load_inventory_stock_items(connection, warehouse_id)
            latest_session_row = self._read_inventory_latest_session_row(connection, warehouse_id)
            latest_inventory = (
                self._serialize_inventory_session_row(latest_session_row)
                if latest_session_row is not None
                else None
            )
            real_item_rows: list[sqlite3.Row] = []
            real_lots_by_item: dict[str, list[dict[str, object]]] = {}
            if latest_session_row is not None:
                real_item_rows, real_lots_by_item = self._load_inventory_session_items(connection, str(latest_session_row["id"]))
            incoming_since_latest = self._sum_inventory_movements_since(
                connection,
                warehouse_id,
                start_value=latest_session_row["created_at"] if latest_session_row is not None else None,
            ) if latest_session_row is not None else 0.0

        current_items = [
            self._serialize_inventory_stock_item_row(row, current_lots_by_item.get(str(row["id"]), []))
            for row in current_item_rows
        ]
        real_items = [
            self._serialize_inventory_stock_item_row(row, real_lots_by_item.get(str(row["id"]), []))
            for row in real_item_rows
        ]
        comparison_items = self._build_inventory_item_comparison(
            current_items=current_items,
            real_items=real_items,
        )
        return {
            "warehouse": self._serialize_inventory_warehouse_row(warehouse_row),
            "items": current_items,
            "current_items": current_items,
            "real_items": real_items,
            "comparison_items": comparison_items,
            "latest_inventory": latest_inventory,
            "total_products": len(current_items),
            "total_equivalent_units": sum(float(item["total_equivalent_units"]) for item in current_items),
            "theoretical_total_equivalent_units": sum(float(item["total_equivalent_units"]) for item in current_items),
            "real_total_equivalent_units": sum(float(item["total_equivalent_units"]) for item in real_items),
            "incoming_since_latest_inventory_equivalent_units": incoming_since_latest,
        }

    def add_inventory_stock(
        self,
        session: SessionIdentity,
        warehouse_id: str,
        payload: InventoryStockWritePayload,
    ) -> dict[str, object]:
        if not self._can_access_inventory(session):
            raise ValueError("Il tuo account non puo accedere a Inventario.")

        timestamp = _iso_now()
        with self._lock:
            with self._connect_tenant_database(session.database_path) as connection:
                warehouse_row = self._read_inventory_warehouse_row(connection, warehouse_id)
                if not self._tenant_table_exists(connection, "ordini_products"):
                    raise ValueError("Catalogo prodotti non disponibile per questo locale.")

                product_columns = self._tenant_table_columns(connection, "ordini_products")
                select_units_per_pack = "units_per_pack" if "units_per_pack" in product_columns else "NULL AS units_per_pack"
                product_row = connection.execute(
                    f"""
                    SELECT id, product_name, lot_code, supplier_name, {select_units_per_pack}
                    FROM ordini_products
                    WHERE id = ? AND active = 1
                    LIMIT 1
                    """,
                    (payload.product_id,),
                ).fetchone()
                if product_row is None:
                    raise KeyError("Prodotto non trovato nel catalogo attivo.")

                product_name = str(product_row["product_name"] or "").strip()
                supplier_name = str(product_row["supplier_name"] or "").strip()
                lot_code = str(product_row["lot_code"] or "").strip()
                product_lookup = _normalize_lookup(product_name)
                supplier_lookup = _normalize_lookup(supplier_name)
                lot_lookup = _normalize_lookup(lot_code)
                units_per_pack = float(product_row["units_per_pack"]) if product_row["units_per_pack"] is not None else None
                operation = str(payload.operation or "add").strip().lower()
                if operation not in {"add", "set"}:
                    raise ValueError("Operazione inventario non valida.")
                if operation == "add" and payload.quantity <= 0:
                    raise ValueError("Per aggiungere stock la quantita deve essere maggiore di zero.")
                if operation == "set" and payload.quantity < 0:
                    raise ValueError("La quantita rilevata non puo essere negativa.")

                target_equivalent_units = self._inventory_effective_units(
                    lot_code=lot_code,
                    quantity=payload.quantity,
                    units_per_pack=units_per_pack,
                )

                item_row = connection.execute(
                    """
                    SELECT *
                    FROM tenant_inventory_stock_items
                    WHERE warehouse_id = ? AND product_lookup = ? AND supplier_lookup = ?
                    LIMIT 1
                    """,
                    (warehouse_id, product_lookup, supplier_lookup),
                ).fetchone()
                current_item_total = float(item_row["total_equivalent_units"] or 0) if item_row is not None else 0.0
                item_id = str(item_row["id"]) if item_row is not None else f"inventory_item_{uuid.uuid4().hex}"

                lot_row = connection.execute(
                    """
                    SELECT *
                    FROM tenant_inventory_stock_lots
                    WHERE item_id = ? AND lot_lookup = ?
                    LIMIT 1
                    """,
                    (item_id, lot_lookup),
                ).fetchone()
                current_lot_quantity = float(lot_row["quantity"] or 0) if lot_row is not None else 0.0
                current_lot_equivalent_units = float(lot_row["equivalent_units"] or 0) if lot_row is not None else 0.0

                if operation == "set":
                    next_lot_quantity = float(payload.quantity)
                    next_lot_equivalent_units = target_equivalent_units
                    delta_equivalent_units = next_lot_equivalent_units - current_lot_equivalent_units
                else:
                    next_lot_quantity = current_lot_quantity + float(payload.quantity)
                    next_lot_equivalent_units = current_lot_equivalent_units + target_equivalent_units
                    delta_equivalent_units = target_equivalent_units

                next_item_total = round(current_item_total + delta_equivalent_units, 6)
                if abs(next_item_total) < 1e-6:
                    next_item_total = 0.0

                if item_row is None and next_lot_quantity <= 0 and next_item_total <= 0:
                    return self.get_inventory_warehouse_detail(session, warehouse_id)

                if item_row is None:
                    item_id = f"inventory_item_{uuid.uuid4().hex}"
                    connection.execute(
                        """
                        INSERT INTO tenant_inventory_stock_items (
                            id,
                            warehouse_id,
                            product_id,
                            product_name,
                            product_lookup,
                            supplier_name,
                            supplier_lookup,
                            total_equivalent_units,
                            created_at,
                            updated_at
                        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                        """,
                        (
                            item_id,
                            warehouse_id,
                            int(product_row["id"]),
                            product_name,
                            product_lookup,
                            supplier_name,
                            supplier_lookup,
                            max(next_item_total, 0.0),
                            timestamp,
                            timestamp,
                        ),
                    )
                else:
                    if next_item_total <= 0:
                        connection.execute(
                            "DELETE FROM tenant_inventory_stock_items WHERE id = ?",
                            (item_id,),
                        )
                    else:
                        connection.execute(
                            """
                            UPDATE tenant_inventory_stock_items
                            SET
                                product_id = ?,
                                product_name = ?,
                                supplier_name = ?,
                                total_equivalent_units = ?,
                                updated_at = ?
                            WHERE id = ?
                            """,
                            (
                                int(product_row["id"]),
                                product_name,
                                supplier_name,
                                max(next_item_total, 0.0),
                                timestamp,
                                item_id,
                            ),
                        )

                if next_item_total <= 0:
                    connection.execute(
                        "DELETE FROM tenant_inventory_stock_lots WHERE item_id = ?",
                        (item_id,),
                    )
                elif lot_row is None:
                    if next_lot_quantity > 0:
                        connection.execute(
                            """
                            INSERT INTO tenant_inventory_stock_lots (
                                id,
                                item_id,
                                lot_code,
                                lot_lookup,
                                quantity,
                                units_per_pack,
                                equivalent_units,
                                created_at,
                                updated_at
                            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                            """,
                            (
                                f"inventory_lot_{uuid.uuid4().hex}",
                                item_id,
                                lot_code,
                                lot_lookup,
                                next_lot_quantity,
                                units_per_pack,
                                next_lot_equivalent_units,
                                timestamp,
                                timestamp,
                            ),
                        )
                else:
                    if next_lot_quantity <= 0 or next_lot_equivalent_units <= 0:
                        connection.execute(
                            "DELETE FROM tenant_inventory_stock_lots WHERE id = ?",
                            (str(lot_row["id"]),),
                        )
                    else:
                        connection.execute(
                            """
                            UPDATE tenant_inventory_stock_lots
                            SET
                                quantity = ?,
                                units_per_pack = COALESCE(?, units_per_pack),
                                equivalent_units = ?,
                                updated_at = ?
                            WHERE id = ?
                            """,
                            (
                                next_lot_quantity,
                                units_per_pack,
                                next_lot_equivalent_units,
                                timestamp,
                                str(lot_row["id"]),
                            ),
                        )

                remaining_lots = connection.execute(
                    "SELECT COUNT(*) AS total FROM tenant_inventory_stock_lots WHERE item_id = ?",
                    (item_id,),
                ).fetchone()
                if remaining_lots is not None and int(remaining_lots["total"] or 0) == 0:
                    connection.execute(
                        "DELETE FROM tenant_inventory_stock_items WHERE id = ?",
                        (item_id,),
                    )

                if operation == "add" and target_equivalent_units > 0:
                    connection.execute(
                        """
                        INSERT INTO tenant_inventory_movements (
                            id,
                            warehouse_id,
                            product_id,
                            product_name,
                            product_lookup,
                            supplier_name,
                            supplier_lookup,
                            lot_code,
                            lot_lookup,
                            quantity,
                            units_per_pack,
                            equivalent_units,
                            movement_kind,
                            source_type,
                            source_label,
                            occurred_at,
                            created_at,
                            updated_at
                        ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                        """,
                        (
                            f"inventory_movement_{uuid.uuid4().hex}",
                            warehouse_id,
                            int(product_row["id"]),
                            product_name,
                            product_lookup,
                            supplier_name,
                            supplier_lookup,
                            lot_code,
                            lot_lookup,
                            float(payload.quantity),
                            units_per_pack,
                            target_equivalent_units,
                            "in",
                            "manual_stock_add",
                            f"Carico manuale {warehouse_row['name']}",
                            timestamp,
                            timestamp,
                            timestamp,
                        ),
                    )

                connection.execute(
                    """
                    UPDATE tenant_inventory_warehouses
                    SET updated_at = ?
                    WHERE id = ?
                    """,
                    (timestamp, warehouse_id),
                )
                connection.commit()

        return self.get_inventory_warehouse_detail(session, warehouse_id)

    def _is_super_admin_configured(self, connection: sqlite3.Connection) -> bool:
        row = connection.execute(
            "SELECT password_hash FROM users WHERE id = ? LIMIT 1",
            (SUPER_ADMIN_USER_ID,),
        ).fetchone()
        return bool(row and row["password_hash"].strip())

    def _build_platform_admin_context(self, session: SessionIdentity) -> dict[str, object] | None:
        if session.role != "super_admin":
            return None

        return {
            "is_super_admin": True,
            "is_impersonating": session.tenant_id != session.home_tenant_id,
            "home_tenant_id": session.home_tenant_id,
            "home_tenant_slug": session.home_tenant_slug,
            "home_tenant_name": session.home_tenant_name,
            "acting_tenant_id": session.tenant_id,
            "acting_tenant_slug": session.tenant_slug,
        }

    def build_auth_response(self, session: SessionIdentity) -> dict[str, object]:
        return {
            "success": True,
            "session_token": session.token,
            "current_user": {
                "id": session.user_id,
                "tenant_id": session.tenant_id,
                "name": session.user_name,
                "username": session.username,
                "email": session.user_email,
                "role": session.role,
                "permissions": list(session.permissions),
            },
            "tenant_context": self.get_tenant_context(session.tenant_id),
            "platform_admin": self._build_platform_admin_context(session),
        }

    def _create_customer_tenant(
        self,
        connection: sqlite3.Connection,
        payload: RegisterTenantPayload | AdminCreateTenantPayload,
    ) -> tuple[str, str]:
        normalized_payload = self._normalize_register_like_payload(payload)
        locale_name = (normalized_payload["locale_name"] or "").strip()
        admin_name = (normalized_payload["admin_name"] or "").strip()
        username = (normalized_payload["username"] or "").strip()
        email = (normalized_payload["email"] or "").strip()
        address = (normalized_payload["address"] or "").strip()
        phone_number = normalized_payload["phone_number"]
        whatsapp_number = normalized_payload["whatsapp_number"]

        if not locale_name or not admin_name or not username or not email or not payload.password.strip():
            raise ValueError("Compila tutti i campi obbligatori per creare il locale")

        slug = _normalize_slug(locale_name)
        tenant_id = f"tenant_{uuid.uuid4().hex}"
        venue_id = f"venue_{uuid.uuid4().hex}"
        user_id = f"user_{uuid.uuid4().hex}"
        created_at = _iso_now()
        password_hash = _hash_password(payload.password.strip())
        database_path = str(self._tenant_dir / f"{slug}.sqlite3")

        slug_exists = connection.execute("SELECT 1 FROM tenants WHERE slug = ?", (slug,)).fetchone()
        if slug_exists is not None:
            raise ValueError("Esiste gia un locale con questo nome. Scegli un nome differente.")

        user_exists = self._lookup_user_for_login(connection, username) or self._lookup_user_for_login(connection, email)
        if user_exists is not None:
            raise ValueError("Username o email gia registrati.")

        connection.execute(
            """
            INSERT INTO tenants (id, name, slug, database_path, created_at)
            VALUES (?, ?, ?, ?, ?)
            """,
            (tenant_id, locale_name, slug, database_path, created_at),
        )
        connection.execute(
            """
            INSERT INTO venues (id, tenant_id, name, address, phone_number, whatsapp_number, created_at)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            """,
            (venue_id, tenant_id, locale_name, address, phone_number, whatsapp_number, created_at),
        )
        connection.execute(
            """
            INSERT INTO users (id, tenant_id, name, username, email, phone_number, password_hash, role, created_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                user_id,
                tenant_id,
                admin_name,
                username,
                email,
                phone_number,
                password_hash,
                "owner",
                created_at,
            ),
        )

        for module in MODULE_CATALOG:
            enabled = 1 if module.status == "active" else 0
            plan_name = "Platform Active" if enabled else "Planned"
            connection.execute(
                """
                INSERT INTO tenant_modules (id, tenant_id, module_key, enabled, plan_name, activated_at, expires_at)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                """,
                (
                    f"tm_{uuid.uuid4().hex}",
                    tenant_id,
                    module.key,
                    enabled,
                    plan_name,
                    created_at if enabled else None,
                    None,
                ),
            )

        connection.commit()
        self._create_tenant_database(
            database_path=database_path,
            tenant_id=tenant_id,
            tenant_name=locale_name,
            tenant_slug=slug,
            venue_name=locale_name,
            address=address,
            phone_number=phone_number,
            whatsapp_number=whatsapp_number,
            admin_name=admin_name,
            admin_username=username,
            admin_email=email,
            admin_phone_number=phone_number,
            password_hash=password_hash,
            created_at=created_at,
        )
        return tenant_id, user_id

    def has_tenants(self) -> bool:
        with self._connect_registry() as connection:
            row = connection.execute(
                "SELECT COUNT(*) AS total FROM tenants WHERE id != ?",
                (SUPER_ADMIN_TENANT_ID,),
            ).fetchone()
            return bool(row and row["total"])

    def status(self) -> dict[str, object]:
        with self._connect_registry() as connection:
            tenant_count = connection.execute(
                "SELECT COUNT(*) AS total FROM tenants WHERE id != ?",
                (SUPER_ADMIN_TENANT_ID,),
            ).fetchone()["total"]
            user_count = connection.execute(
                "SELECT COUNT(*) AS total FROM users WHERE id != ?",
                (SUPER_ADMIN_USER_ID,),
            ).fetchone()["total"]
            super_admin_configured = self._is_super_admin_configured(connection)
            return {
                "status": "ready",
                "mode": "sqlite-multi-tenant",
                "tenant_count": tenant_count,
                "user_count": user_count,
                "first_run": not super_admin_configured,
                "super_admin_bootstrap_required": not super_admin_configured,
                "super_admin_identifier": SUPER_ADMIN_USERNAME,
                "next_step": (
                    "Imposta la password iniziale del super admin."
                    if not super_admin_configured
                    else "Accedi come super admin e crea o gestisci i locali dal pannello admin."
                ),
            }

    def register(self, payload: RegisterTenantPayload) -> SessionIdentity:
        with self._lock:
            with self._connect_registry() as connection:
                tenant_id, user_id = self._create_customer_tenant(connection, payload)
                return self._create_session(connection, tenant_id=tenant_id, user_id=user_id)

    def login(self, payload: LoginPayload) -> SessionIdentity:
        identifier = payload.identifier.strip()
        password = payload.password.strip()
        if not identifier or not password:
            raise ValueError("Inserisci credenziali valide.")

        with self._lock:
            with self._connect_registry() as connection:
                candidates = self._lookup_login_candidates(connection, identifier)
                if not candidates:
                    raise ValueError("Credenziali non valide.")

                bootstrap_required = False
                for user in candidates:
                    stored_hash = (user["password_hash"] or "").strip()
                    if user["role"] == "super_admin" and not stored_hash:
                        bootstrap_required = True
                        continue
                    if not stored_hash or not _verify_password(password, stored_hash):
                        continue
                    return self._create_session(connection, tenant_id=user["tenant_id"], user_id=user["id"])

                if bootstrap_required and len(candidates) == 1 and candidates[0]["role"] == "super_admin":
                    raise ValueError("Password super admin non ancora impostata. Completa la prima attivazione.")

                raise ValueError("Credenziali non valide.")

    def logout(self, token: str) -> None:
        with self._lock:
            with self._connect_registry() as connection:
                connection.execute("DELETE FROM sessions WHERE token = ?", (token,))
                connection.commit()

    def get_session(self, token: str) -> SessionIdentity | None:
        with self._connect_registry() as connection:
            row = connection.execute(
                """
                SELECT
                    sessions.token,
                    sessions.tenant_id,
                    sessions.user_id,
                    sessions.expires_at,
                    tenants.slug AS tenant_slug,
                    tenants.name AS tenant_name,
                    home_tenants.id AS home_tenant_id,
                    home_tenants.slug AS home_tenant_slug,
                    home_tenants.name AS home_tenant_name,
                    tenants.database_path AS database_path,
                    users.email AS user_email,
                    users.username AS username,
                    users.name AS user_name,
                    users.role AS role,
                    users.permissions_json AS permissions_json
                FROM sessions
                JOIN tenants ON tenants.id = sessions.tenant_id
                JOIN users ON users.id = sessions.user_id
                JOIN tenants AS home_tenants ON home_tenants.id = users.tenant_id
                WHERE sessions.token = ?
                LIMIT 1
                """,
                (token,),
            ).fetchone()

            if row is None:
                return None

            expires_at = datetime.fromisoformat(row["expires_at"])
            if expires_at < _utcnow():
                connection.execute("DELETE FROM sessions WHERE token = ?", (token,))
                connection.commit()
                return None

            return self._session_identity_from_row(row)

    def build_service_session(self, tenant_id: str) -> SessionIdentity | None:
        with self._connect_registry() as connection:
            row = connection.execute(
                """
                SELECT
                    tenants.id AS tenant_id,
                    tenants.slug AS tenant_slug,
                    tenants.name AS tenant_name,
                    tenants.database_path AS database_path,
                    users.id AS user_id,
                    users.email AS user_email,
                    users.username AS username,
                    users.name AS user_name,
                    users.role AS role,
                    users.permissions_json AS permissions_json
                FROM tenants
                JOIN users ON users.tenant_id = tenants.id
                WHERE tenants.id = ?
                ORDER BY
                    CASE users.role
                        WHEN 'owner' THEN 0
                        WHEN 'staff' THEN 1
                        ELSE 2
                    END,
                    users.created_at ASC
                LIMIT 1
                """,
                (tenant_id,),
            ).fetchone()

        if row is None:
            return None

        return SessionIdentity(
            token=f"service:{tenant_id}",
            tenant_id=row["tenant_id"],
            tenant_slug=row["tenant_slug"],
            tenant_name=row["tenant_name"],
            home_tenant_id=row["tenant_id"],
            home_tenant_slug=row["tenant_slug"],
            home_tenant_name=row["tenant_name"],
            user_id=row["user_id"],
            user_email=row["user_email"],
            username=row["username"],
            user_name=row["user_name"],
            role=row["role"],
            permissions=_resolve_role_permissions(row["role"], row["permissions_json"]),
            database_path=row["database_path"],
            expires_at=(_utcnow() + timedelta(days=3650)).isoformat(),
        )

    def get_tenant_context(self, tenant_id: str) -> dict[str, object]:
        with self._connect_registry() as connection:
            tenant_row, venue_rows, user_rows, module_rows = self._context_rows(connection, tenant_id)

        return {
            "tenant": Tenant(
                id=tenant_row["id"],
                name=tenant_row["name"],
                slug=tenant_row["slug"],
                created_at=tenant_row["created_at"],
            ),
            "venues": [
                Venue(
                    id=row["id"],
                    tenant_id=row["tenant_id"],
                    name=row["name"],
                    address=row["address"],
                    phone_number=row["phone_number"],
                    whatsapp_number=row["whatsapp_number"],
                    created_at=row["created_at"],
                )
                for row in venue_rows
            ],
            "users": [
                User(
                    id=row["id"],
                    tenant_id=row["tenant_id"],
                    name=row["name"],
                    username=row["username"],
                    email=row["email"],
                    phone_number=row["phone_number"],
                    role=row["role"],
                    created_at=row["created_at"],
                )
                for row in user_rows
            ],
            "tenant_modules": [
                TenantModule(
                    id=row["id"],
                    tenant_id=row["tenant_id"],
                    module_key=row["module_key"],
                    enabled=bool(row["enabled"]),
                    plan_name=row["plan_name"],
                    activated_at=row["activated_at"],
                    expires_at=row["expires_at"],
                )
                for row in module_rows
            ],
        }

    def update_venue_info(
        self,
        session: SessionIdentity,
        *,
        venue_name: str | None = None,
        address: str | None = None,
        phone_number: str | None = None,
        whatsapp_number: str | None = None,
    ) -> None:
        if session.role not in ("owner", "manager", "super_admin"):
            raise ValueError("Operazione riservata all'admin del locale.")

        registry_assignments: list[str] = []
        registry_values: list[object] = []
        tenant_assignments: list[str] = []
        tenant_values: list[object] = []

        if venue_name is not None:
            v = venue_name.strip() or None
            registry_assignments.append("name = ?")
            registry_values.append(v)
            tenant_assignments.append("venue_name = ?")
            tenant_values.append(v)
        if address is not None:
            v = address.strip() or None
            registry_assignments.append("address = ?")
            registry_values.append(v)
            tenant_assignments.append("address = ?")
            tenant_values.append(v)
        if phone_number is not None:
            v = phone_number.strip() or None
            registry_assignments.append("phone_number = ?")
            registry_values.append(v)
            tenant_assignments.append("phone_number = ?")
            tenant_values.append(v)
        if whatsapp_number is not None:
            v = whatsapp_number.strip() or None
            registry_assignments.append("whatsapp_number = ?")
            registry_values.append(v)
            tenant_assignments.append("whatsapp_number = ?")
            tenant_values.append(v)

        if not registry_assignments:
            return

        with self._lock:
            with self._connect_registry() as connection:
                venue_row = connection.execute(
                    "SELECT id FROM venues WHERE tenant_id = ? LIMIT 1",
                    (session.tenant_id,),
                ).fetchone()
                if venue_row is None:
                    raise KeyError("Nessun locale trovato per questo tenant.")
                registry_values.append(venue_row["id"])
                connection.execute(
                    f"UPDATE venues SET {', '.join(registry_assignments)} WHERE id = ?",
                    registry_values,
                )
                connection.commit()

        if session.database_path and tenant_assignments:
            with self._connect_tenant_database(session.database_path) as tenant_conn:
                tenant_values.append(session.tenant_id)
                tenant_conn.execute(
                    f"UPDATE tenant_profile SET {', '.join(tenant_assignments)} WHERE tenant_id = ?",
                    tenant_values,
                )
                tenant_conn.commit()

    def get_context_by_slug(self, tenant_slug: str) -> dict[str, object] | None:
        with self._connect_registry() as connection:
            tenant = connection.execute("SELECT id FROM tenants WHERE slug = ? LIMIT 1", (tenant_slug,)).fetchone()
            if tenant is None:
                return None

        return self.get_tenant_context(tenant["id"])

    def bootstrap_super_admin(self, payload: BootstrapSuperAdminPayload) -> SessionIdentity:
        password = payload.password.strip()
        if not password:
            raise ValueError("Inserisci una password valida per il super admin.")

        with self._lock:
            with self._connect_registry() as connection:
                if self._is_super_admin_configured(connection):
                    raise ValueError("Il super admin e gia stato configurato.")

                password_hash = _hash_password(password)
                connection.execute(
                    "UPDATE users SET password_hash = ? WHERE id = ?",
                    (password_hash, SUPER_ADMIN_USER_ID),
                )
                connection.commit()
                session = self._create_session(connection, tenant_id=SUPER_ADMIN_TENANT_ID, user_id=SUPER_ADMIN_USER_ID)

        admin_db_path = str(self._tenant_dir / f"{SUPER_ADMIN_TENANT_SLUG}.sqlite3")
        with self._connect_tenant_database(admin_db_path) as admin_connection:
            profile = admin_connection.execute(
                "SELECT created_at FROM tenant_profile WHERE tenant_id = ? LIMIT 1",
                (SUPER_ADMIN_TENANT_ID,),
            ).fetchone()
            created_at = profile["created_at"] if profile else _iso_now()
            self._upsert_tenant_profile(
                admin_connection,
                tenant_id=SUPER_ADMIN_TENANT_ID,
                tenant_name=SUPER_ADMIN_TENANT_NAME,
                tenant_slug=SUPER_ADMIN_TENANT_SLUG,
                venue_name=SUPER_ADMIN_TENANT_NAME,
                address="",
                phone_number=None,
                whatsapp_number=None,
                admin_name="Super Admin",
                admin_username=SUPER_ADMIN_USERNAME,
                admin_email=SUPER_ADMIN_EMAIL,
                admin_phone_number=None,
                password_hash=password_hash,
                created_at=created_at,
            )
            admin_connection.commit()

        return session

    def update_super_admin_password(self, session: SessionIdentity, payload: UpdatePasswordPayload) -> None:
        if session.role != "super_admin":
            raise ValueError("Operazione riservata al super admin.")

        new_password = payload.new_password.strip()
        current_password = payload.current_password.strip() if payload.current_password else ""
        if not new_password:
            raise ValueError("Inserisci una nuova password valida.")

        with self._lock:
            with self._connect_registry() as connection:
                row = connection.execute(
                    "SELECT password_hash FROM users WHERE id = ? LIMIT 1",
                    (SUPER_ADMIN_USER_ID,),
                ).fetchone()
                if row is None:
                    raise ValueError("Account super admin non disponibile.")

                stored_hash = (row["password_hash"] or "").strip()
                if stored_hash and not current_password:
                    raise ValueError("Inserisci la password attuale del super admin.")
                if stored_hash and not _verify_password(current_password, stored_hash):
                    raise ValueError("La password attuale non e corretta.")

                new_hash = _hash_password(new_password)
                connection.execute(
                    "UPDATE users SET password_hash = ? WHERE id = ?",
                    (new_hash, SUPER_ADMIN_USER_ID),
                )
                connection.commit()

        admin_db_path = str(self._tenant_dir / f"{SUPER_ADMIN_TENANT_SLUG}.sqlite3")
        with self._connect_tenant_database(admin_db_path) as admin_connection:
            profile = admin_connection.execute(
                "SELECT created_at FROM tenant_profile WHERE tenant_id = ? LIMIT 1",
                (SUPER_ADMIN_TENANT_ID,),
            ).fetchone()
            created_at = profile["created_at"] if profile else _iso_now()
            self._upsert_tenant_profile(
                admin_connection,
                tenant_id=SUPER_ADMIN_TENANT_ID,
                tenant_name=SUPER_ADMIN_TENANT_NAME,
                tenant_slug=SUPER_ADMIN_TENANT_SLUG,
                venue_name=SUPER_ADMIN_TENANT_NAME,
                address="",
                phone_number=None,
                whatsapp_number=None,
                admin_name=session.user_name or "Super Admin",
                admin_username=SUPER_ADMIN_USERNAME,
                admin_email=SUPER_ADMIN_EMAIL,
                admin_phone_number=None,
                password_hash=new_hash,
                created_at=created_at,
            )
            admin_connection.commit()

    def update_current_user_password(self, session: SessionIdentity, payload: UpdatePasswordPayload) -> None:
        if session.role == "super_admin":
            self.update_super_admin_password(session, payload)
            return

        new_password = payload.new_password.strip()
        current_password = payload.current_password.strip() if payload.current_password else ""
        if not new_password:
            raise ValueError("Inserisci una nuova password valida.")

        with self._lock:
            with self._connect_registry() as connection:
                row = connection.execute(
                    """
                    SELECT password_hash
                    FROM users
                    WHERE id = ? AND tenant_id = ?
                    LIMIT 1
                    """,
                    (session.user_id, session.tenant_id),
                ).fetchone()
                if row is None:
                    raise ValueError("Account del locale non disponibile.")

                stored_hash = (row["password_hash"] or "").strip()
                if stored_hash and not current_password:
                    raise ValueError("Inserisci la password attuale.")
                if stored_hash and not _verify_password(current_password, stored_hash):
                    raise ValueError("La password attuale non e corretta.")

                new_hash = _hash_password(new_password)
                connection.execute(
                    "UPDATE users SET password_hash = ? WHERE id = ?",
                    (new_hash, session.user_id),
                )
                connection.commit()

        if session.role == "owner":
            with self._connect_tenant_database(session.database_path) as tenant_connection:
                tenant_connection.execute(
                    """
                    UPDATE tenant_profile
                    SET password_hash = ?, updated_at = ?
                    WHERE tenant_id = ?
                    """,
                    (new_hash, _iso_now(), session.tenant_id),
                )
                tenant_connection.commit()

    def _tenant_summary_from_row(self, row: sqlite3.Row) -> dict[str, object]:
        venue_payload = None
        if row["venue_id"] is not None:
            venue_payload = Venue(
                id=row["venue_id"],
                tenant_id=row["id"],
                name=row["venue_name"],
                address=row["venue_address"],
                phone_number=row["venue_phone_number"],
                whatsapp_number=row["venue_whatsapp_number"],
                created_at=row["venue_created_at"],
            )

        admin_payload = None
        if row["admin_user_id"] is not None:
            admin_payload = User(
                id=row["admin_user_id"],
                tenant_id=row["id"],
                name=row["admin_user_name"],
                username=row["admin_username"],
                email=row["admin_email"],
                phone_number=row["admin_phone_number"],
                role=row["admin_role"],
                created_at=row["admin_created_at"],
            )

        return {
            "tenant": Tenant(
                id=row["id"],
                name=row["name"],
                slug=row["slug"],
                created_at=row["created_at"],
            ),
            "venue": venue_payload,
            "admin_user": admin_payload,
            "enabled_modules": int(row["enabled_modules"] or 0),
            "total_modules": int(row["total_modules"] or 0),
        }

    def list_customer_tenants(self) -> list[dict[str, object]]:
        with self._connect_registry() as connection:
            rows = list(
                connection.execute(
                    """
                    SELECT
                        tenants.*,
                        venues.id AS venue_id,
                        venues.name AS venue_name,
                        venues.address AS venue_address,
                        venues.phone_number AS venue_phone_number,
                        venues.whatsapp_number AS venue_whatsapp_number,
                        venues.created_at AS venue_created_at,
                        users.id AS admin_user_id,
                        users.name AS admin_user_name,
                        users.username AS admin_username,
                        users.email AS admin_email,
                        users.phone_number AS admin_phone_number,
                        users.role AS admin_role,
                        users.created_at AS admin_created_at,
                        COALESCE(module_stats.enabled_modules, 0) AS enabled_modules,
                        COALESCE(module_stats.total_modules, 0) AS total_modules
                    FROM tenants
                    LEFT JOIN venues ON venues.id = (
                        SELECT id
                        FROM venues AS first_venues
                        WHERE first_venues.tenant_id = tenants.id
                        ORDER BY first_venues.created_at ASC
                        LIMIT 1
                    )
                    LEFT JOIN users ON users.id = (
                        SELECT id
                        FROM users AS first_users
                        WHERE first_users.tenant_id = tenants.id
                        ORDER BY first_users.created_at ASC
                        LIMIT 1
                    )
                    LEFT JOIN (
                        SELECT
                            tenant_id,
                            SUM(CASE WHEN enabled = 1 THEN 1 ELSE 0 END) AS enabled_modules,
                            COUNT(*) AS total_modules
                        FROM tenant_modules
                        GROUP BY tenant_id
                    ) AS module_stats ON module_stats.tenant_id = tenants.id
                    WHERE tenants.id != ?
                    ORDER BY tenants.created_at DESC
                    """,
                    (SUPER_ADMIN_TENANT_ID,),
                )
            )

        return [self._tenant_summary_from_row(row) for row in rows]

    def get_customer_tenant_summary(self, tenant_id: str) -> dict[str, object]:
        with self._connect_registry() as connection:
            row = connection.execute(
                """
                SELECT
                    tenants.*,
                    venues.id AS venue_id,
                    venues.name AS venue_name,
                    venues.address AS venue_address,
                    venues.phone_number AS venue_phone_number,
                    venues.whatsapp_number AS venue_whatsapp_number,
                    venues.created_at AS venue_created_at,
                    users.id AS admin_user_id,
                    users.name AS admin_user_name,
                    users.username AS admin_username,
                    users.email AS admin_email,
                    users.phone_number AS admin_phone_number,
                    users.role AS admin_role,
                    users.created_at AS admin_created_at,
                    COALESCE(module_stats.enabled_modules, 0) AS enabled_modules,
                    COALESCE(module_stats.total_modules, 0) AS total_modules
                FROM tenants
                LEFT JOIN venues ON venues.id = (
                    SELECT id
                    FROM venues AS first_venues
                    WHERE first_venues.tenant_id = tenants.id
                    ORDER BY first_venues.created_at ASC
                    LIMIT 1
                )
                LEFT JOIN users ON users.id = (
                    SELECT id
                    FROM users AS first_users
                    WHERE first_users.tenant_id = tenants.id
                    ORDER BY first_users.created_at ASC
                    LIMIT 1
                )
                LEFT JOIN (
                    SELECT
                        tenant_id,
                        SUM(CASE WHEN enabled = 1 THEN 1 ELSE 0 END) AS enabled_modules,
                        COUNT(*) AS total_modules
                    FROM tenant_modules
                    GROUP BY tenant_id
                ) AS module_stats ON module_stats.tenant_id = tenants.id
                WHERE tenants.id = ? AND tenants.id != ?
                LIMIT 1
                """,
                (tenant_id, SUPER_ADMIN_TENANT_ID),
            ).fetchone()

        if row is None:
            raise KeyError(f"Tenant {tenant_id} non trovato")
        return self._tenant_summary_from_row(row)

    def get_admin_overview(self) -> dict[str, object]:
        with self._connect_registry() as connection:
            super_admin = connection.execute(
                """
                SELECT id, name, username, email, role, created_at
                FROM users
                WHERE id = ?
                LIMIT 1
                """,
                (SUPER_ADMIN_USER_ID,),
            ).fetchone()
            password_configured = self._is_super_admin_configured(connection)

        if super_admin is None:
            raise KeyError("Account super admin non trovato")

        return {
            "super_admin": {
                "id": super_admin["id"],
                "name": super_admin["name"],
                "username": super_admin["username"],
                "email": super_admin["email"],
                "role": super_admin["role"],
                "created_at": super_admin["created_at"],
                "password_configured": password_configured,
            },
            "tenants": self.list_customer_tenants(),
        }

    def create_tenant_as_super_admin(self, payload: AdminCreateTenantPayload) -> dict[str, object]:
        with self._lock:
            with self._connect_registry() as connection:
                tenant_id, _user_id = self._create_customer_tenant(connection, payload)

        return self.get_customer_tenant_summary(tenant_id)

    def update_tenant_admin(self, tenant_id: str, payload: AdminUpdateTenantAdminPayload) -> dict[str, object]:
        normalized_payload = self._normalize_register_like_payload(payload)
        locale_name = (normalized_payload["locale_name"] or "").strip()
        admin_name = (normalized_payload["admin_name"] or "").strip()
        username = (normalized_payload["username"] or "").strip()
        email = (normalized_payload["email"] or "").strip()
        address = (normalized_payload["address"] or "").strip()
        phone_number = normalized_payload["phone_number"]
        whatsapp_number = normalized_payload["whatsapp_number"]
        next_password = payload.password.strip() if payload.password else None

        if tenant_id == SUPER_ADMIN_TENANT_ID:
            raise ValueError("Il tenant di piattaforma non puo essere modificato da questo endpoint.")
        if not locale_name or not admin_name or not username or not email:
            raise ValueError("Compila tutti i campi obbligatori del locale.")

        with self._lock:
            with self._connect_registry() as connection:
                tenant_row = connection.execute("SELECT * FROM tenants WHERE id = ? LIMIT 1", (tenant_id,)).fetchone()
                if tenant_row is None:
                    raise KeyError(f"Tenant {tenant_id} non trovato")

                venue_row = connection.execute(
                    "SELECT * FROM venues WHERE tenant_id = ? ORDER BY created_at ASC LIMIT 1",
                    (tenant_id,),
                ).fetchone()
                user_row = connection.execute(
                    "SELECT * FROM users WHERE tenant_id = ? ORDER BY created_at ASC LIMIT 1",
                    (tenant_id,),
                ).fetchone()
                if venue_row is None or user_row is None:
                    raise ValueError("Tenant incompleto: venue o admin non trovati.")

                username_conflict = connection.execute(
                    "SELECT id FROM users WHERE lower(username) = ? AND id != ? LIMIT 1",
                    (_normalize_lookup(username), user_row["id"]),
                ).fetchone()
                email_conflict = connection.execute(
                    "SELECT id FROM users WHERE lower(email) = ? AND id != ? LIMIT 1",
                    (_normalize_lookup(email), user_row["id"]),
                ).fetchone()
                if username_conflict is not None or email_conflict is not None:
                    raise ValueError("Username o email gia registrati da un altro account.")

                connection.execute(
                    "UPDATE tenants SET name = ? WHERE id = ?",
                    (locale_name, tenant_id),
                )
                connection.execute(
                    """
                    UPDATE venues
                    SET name = ?, address = ?, phone_number = ?, whatsapp_number = ?
                    WHERE id = ?
                    """,
                    (locale_name, address, phone_number, whatsapp_number, venue_row["id"]),
                )

                assignments = [
                    "name = ?",
                    "username = ?",
                    "email = ?",
                    "phone_number = ?",
                ]
                values: list[object] = [admin_name, username, email, phone_number]
                password_hash = user_row["password_hash"]
                if next_password:
                    password_hash = _hash_password(next_password)
                    assignments.append("password_hash = ?")
                    values.append(password_hash)

                values.append(user_row["id"])
                connection.execute(
                    f"UPDATE users SET {', '.join(assignments)} WHERE id = ?",
                    values,
                )
                connection.commit()

                self._create_tenant_database(
                    database_path=tenant_row["database_path"],
                    tenant_id=tenant_id,
                    tenant_name=locale_name,
                    tenant_slug=tenant_row["slug"],
                    venue_name=locale_name,
                    address=address,
                    phone_number=phone_number,
                    whatsapp_number=whatsapp_number,
                    admin_name=admin_name,
                    admin_username=username,
                    admin_email=email,
                    admin_phone_number=phone_number,
                    password_hash=password_hash,
                    created_at=tenant_row["created_at"],
                )

        return self.get_customer_tenant_summary(tenant_id)

    def list_tenant_users(self, session: SessionIdentity) -> list[dict[str, object]]:
        if not self._can_manage_tenant_accounts(session):
            raise ValueError("Operazione riservata all'admin del locale.")

        with self._connect_registry() as connection:
            rows = list(
                connection.execute(
                    """
                    SELECT id, tenant_id, name, username, email, phone_number, role, permissions_json, created_at
                    FROM users
                    WHERE tenant_id = ?
                    ORDER BY
                        CASE role
                            WHEN 'owner' THEN 0
                            WHEN 'manager' THEN 1
                            WHEN 'staff' THEN 2
                            ELSE 3
                        END,
                        created_at ASC
                    """,
                    (session.tenant_id,),
                )
            )

        return [
            {
                "id": row["id"],
                "tenant_id": row["tenant_id"],
                "name": row["name"],
                "username": row["username"],
                "email": row["email"],
                "phone_number": row["phone_number"],
                "role": row["role"],
                "permissions": list(_resolve_role_permissions(row["role"], row["permissions_json"])),
                "created_at": row["created_at"],
            }
            for row in rows
        ]

    def create_tenant_staff_user(self, session: SessionIdentity, payload: TenantStaffUserCreatePayload) -> dict[str, object]:
        if not self._can_manage_tenant_accounts(session):
            raise ValueError("Operazione riservata all'admin del locale.")

        normalized_payload = self._normalize_tenant_staff_payload(payload)
        name = (normalized_payload["name"] or "").strip()
        username = (normalized_payload["username"] or "").strip()
        email = (normalized_payload["email"] or "").strip()
        phone_number = normalized_payload["phone_number"]
        password = payload.password.strip()
        permissions_json = _serialize_role_permissions("staff", payload.permissions)

        if not name or not username or not email or not password:
            raise ValueError("Compila tutti i campi obbligatori del sotto-account.")

        with self._lock:
            with self._connect_registry() as connection:
                username_conflict = connection.execute(
                    "SELECT id FROM users WHERE lower(username) = ? LIMIT 1",
                    (_normalize_lookup(username),),
                ).fetchone()
                email_conflict = connection.execute(
                    "SELECT id FROM users WHERE lower(email) = ? LIMIT 1",
                    (_normalize_lookup(email),),
                ).fetchone()
                if username_conflict is not None or email_conflict is not None:
                    raise ValueError("Username o email gia registrati da un altro account.")

                user_id = f"user_{uuid.uuid4().hex}"
                created_at = _iso_now()
                connection.execute(
                    """
                    INSERT INTO users (id, tenant_id, name, username, email, phone_number, password_hash, role, permissions_json, created_at)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    (
                        user_id,
                        session.tenant_id,
                        name,
                        username,
                        email,
                        phone_number,
                        _hash_password(password),
                        "staff",
                        permissions_json,
                        created_at,
                    ),
                )
                connection.commit()

        return next(
            user
            for user in self.list_tenant_users(session)
            if user["id"] == user_id
        )

    def update_tenant_staff_user(
        self,
        session: SessionIdentity,
        user_id: str,
        payload: TenantStaffUserUpdatePayload,
    ) -> dict[str, object]:
        if not self._can_manage_tenant_accounts(session):
            raise ValueError("Operazione riservata all'admin del locale.")

        normalized_payload = self._normalize_tenant_staff_payload(payload)
        name = (normalized_payload["name"] or "").strip()
        username = (normalized_payload["username"] or "").strip()
        email = (normalized_payload["email"] or "").strip()
        phone_number = normalized_payload["phone_number"]
        next_password = payload.password.strip() if payload.password else None
        permissions_json = _serialize_role_permissions("staff", payload.permissions)

        if not user_id or not name or not username or not email:
            raise ValueError("Compila tutti i campi obbligatori del sotto-account.")

        with self._lock:
            with self._connect_registry() as connection:
                user_row = connection.execute(
                    """
                    SELECT * FROM users
                    WHERE id = ? AND tenant_id = ?
                    LIMIT 1
                    """,
                    (user_id, session.tenant_id),
                ).fetchone()
                if user_row is None:
                    raise KeyError("Sotto-account non trovato.")
                if user_row["role"] != "staff":
                    raise ValueError("Puoi modificare solo i sotto-account del locale.")

                username_conflict = connection.execute(
                    "SELECT id FROM users WHERE lower(username) = ? AND id != ? LIMIT 1",
                    (_normalize_lookup(username), user_id),
                ).fetchone()
                email_conflict = connection.execute(
                    "SELECT id FROM users WHERE lower(email) = ? AND id != ? LIMIT 1",
                    (_normalize_lookup(email), user_id),
                ).fetchone()
                if username_conflict is not None or email_conflict is not None:
                    raise ValueError("Username o email gia registrati da un altro account.")

                assignments = [
                    "name = ?",
                    "username = ?",
                    "email = ?",
                    "phone_number = ?",
                    "permissions_json = ?",
                ]
                values: list[object] = [name, username, email, phone_number, permissions_json]
                if next_password:
                    assignments.append("password_hash = ?")
                    values.append(_hash_password(next_password))

                values.append(user_id)
                connection.execute(
                    f"UPDATE users SET {', '.join(assignments)} WHERE id = ?",
                    values,
                )
                connection.commit()

        return next(
            user
            for user in self.list_tenant_users(session)
            if user["id"] == user_id
        )

    def delete_tenant_staff_user(self, session: SessionIdentity, user_id: str) -> None:
        if not self._can_manage_tenant_accounts(session):
            raise ValueError("Operazione riservata all'admin del locale.")

        if session.user_id == user_id:
            raise ValueError("Non puoi eliminare l'account con cui sei autenticato.")

        with self._lock:
            with self._connect_registry() as connection:
                user_row = connection.execute(
                    """
                    SELECT * FROM users
                    WHERE id = ? AND tenant_id = ?
                    LIMIT 1
                    """,
                    (user_id, session.tenant_id),
                ).fetchone()
                if user_row is None:
                    raise KeyError("Sotto-account non trovato.")
                if user_row["role"] != "staff":
                    raise ValueError("Puoi eliminare solo i sotto-account del locale.")

                connection.execute(
                    "DELETE FROM assistant_runs WHERE tenant_id = ? AND user_id = ?",
                    (session.tenant_id, user_id),
                )
                connection.execute(
                    "DELETE FROM assistant_threads WHERE tenant_id = ? AND user_id = ?",
                    (session.tenant_id, user_id),
                )
                connection.execute("DELETE FROM sessions WHERE user_id = ?", (user_id,))
                connection.execute("DELETE FROM users WHERE id = ?", (user_id,))
                connection.commit()
            with self._connect_tenant_database(session.database_path) as tenant_connection:
                tenant_connection.execute("DELETE FROM tenant_timeclock_entries WHERE user_id = ?", (user_id,))
                tenant_connection.commit()

    def get_timeclock_status(self, session: SessionIdentity) -> dict[str, object]:
        if not self._can_access_timeclock(session):
            raise ValueError("Questo account non puo accedere a Cartellino.")

        now = _utcnow()
        today = now.date().isoformat()
        week_start = (now.date() - timedelta(days=now.weekday())).isoformat()

        recent_entries = self._read_timeclock_entries(session, user_id=session.user_id, limit=20)
        active_entry = next((entry for entry in recent_entries if entry.ended_at is None), None)
        today_entries = self._read_timeclock_entries(session, user_id=session.user_id, start_date=today, end_date=today, limit=500)
        week_entries = self._read_timeclock_entries(
            session,
            user_id=session.user_id,
            start_date=week_start,
            end_date=today,
            limit=1000,
        )

        today_seconds = sum(int(self._serialize_timeclock_entry(entry, reference_now=now)["duration_seconds"] or 0) for entry in today_entries)
        week_seconds = sum(int(self._serialize_timeclock_entry(entry, reference_now=now)["duration_seconds"] or 0) for entry in week_entries)

        return {
            "can_manage_team": session.role in {"owner", "super_admin"},
            "active_entry": self._serialize_timeclock_entry(active_entry, reference_now=now) if active_entry is not None else None,
            "today_seconds": today_seconds,
            "today_hours": round(today_seconds / 3600, 2),
            "week_seconds": week_seconds,
            "week_hours": round(week_seconds / 3600, 2),
            "recent_entries": [self._serialize_timeclock_entry(entry, reference_now=now) for entry in recent_entries],
        }

    def start_timeclock_shift(self, session: SessionIdentity) -> dict[str, object]:
        if not self._can_access_timeclock(session):
            raise ValueError("Questo account non puo accedere a Cartellino.")

        started_at = _iso_now()
        with self._lock:
            with self._connect_tenant_database(session.database_path) as connection:
                active_row = connection.execute(
                    """
                    SELECT *
                    FROM tenant_timeclock_entries
                    WHERE user_id = ? AND ended_at IS NULL
                    ORDER BY started_at DESC
                    LIMIT 1
                    """,
                    (session.user_id,),
                ).fetchone()
                if active_row is not None:
                    raise ValueError("Esiste gia un turno attivo per questo account.")

                entry_id = f"timeclock_{uuid.uuid4().hex}"
                connection.execute(
                    """
                    INSERT INTO tenant_timeclock_entries (
                        id,
                        user_id,
                        user_name,
                        username,
                        user_email,
                        started_at,
                        ended_at,
                        duration_seconds,
                        started_source,
                        ended_source,
                        notes,
                        created_at,
                        updated_at
                    ) VALUES (?, ?, ?, ?, ?, ?, NULL, NULL, 'portal', NULL, NULL, ?, ?)
                    """,
                    (
                        entry_id,
                        session.user_id,
                        session.user_name,
                        session.username,
                        session.user_email,
                        started_at,
                        started_at,
                        started_at,
                    ),
                )
                connection.commit()
                row = connection.execute(
                    "SELECT * FROM tenant_timeclock_entries WHERE id = ? LIMIT 1",
                    (entry_id,),
                ).fetchone()

        if row is None:
            raise ValueError("Turno avviato ma non piu disponibile.")
        return {"entry": self._serialize_timeclock_entry(self._timeclock_entry_from_row(row))}

    def stop_timeclock_shift(self, session: SessionIdentity) -> dict[str, object]:
        if not self._can_access_timeclock(session):
            raise ValueError("Questo account non puo accedere a Cartellino.")

        stopped_at = _utcnow()
        stopped_at_iso = stopped_at.isoformat()
        with self._lock:
            with self._connect_tenant_database(session.database_path) as connection:
                active_row = connection.execute(
                    """
                    SELECT *
                    FROM tenant_timeclock_entries
                    WHERE user_id = ? AND ended_at IS NULL
                    ORDER BY started_at DESC
                    LIMIT 1
                    """,
                    (session.user_id,),
                ).fetchone()
                if active_row is None:
                    raise ValueError("Non vedo un turno attivo da chiudere per questo account.")

                active_entry = self._timeclock_entry_from_row(active_row)
                started_at = self._parse_iso_datetime(active_entry.started_at)
                duration_seconds = 0
                if started_at is not None:
                    duration_seconds = max(int((stopped_at - started_at).total_seconds()), 0)

                connection.execute(
                    """
                    UPDATE tenant_timeclock_entries
                    SET ended_at = ?, duration_seconds = ?, ended_source = 'portal', updated_at = ?
                    WHERE id = ?
                    """,
                    (stopped_at_iso, duration_seconds, stopped_at_iso, active_entry.id),
                )
                connection.commit()
                row = connection.execute(
                    "SELECT * FROM tenant_timeclock_entries WHERE id = ? LIMIT 1",
                    (active_entry.id,),
                ).fetchone()

        if row is None:
            raise ValueError("Turno chiuso ma non piu disponibile.")
        return {"entry": self._serialize_timeclock_entry(self._timeclock_entry_from_row(row))}

    def get_timeclock_overview(self, session: SessionIdentity, query: TimeclockOverviewQuery) -> dict[str, object]:
        if not self._can_access_timeclock(session):
            raise ValueError("Questo account non puo accedere a Cartellino.")

        effective_user_id = query.user_id.strip() if query.user_id else None
        can_manage_team = session.role in {"owner", "super_admin"}
        if not can_manage_team:
            effective_user_id = session.user_id

        now = _utcnow()
        entries = self._read_timeclock_entries(
            session,
            user_id=effective_user_id,
            start_date=query.start_date.strip() if query.start_date else None,
            end_date=query.end_date.strip() if query.end_date else None,
            limit=query.limit,
        )
        active_entries = self._read_timeclock_entries(
            session,
            user_id=effective_user_id,
            include_active_only=True,
            limit=200,
        )
        if not can_manage_team:
            active_entries = [entry for entry in active_entries if entry.user_id == session.user_id]

        return {
            "can_manage_team": can_manage_team,
            "filters": {
                "user_id": effective_user_id,
                "start_date": query.start_date,
                "end_date": query.end_date,
                "limit": query.limit,
            },
            "entries": [self._serialize_timeclock_entry(entry, reference_now=now) for entry in entries],
            "summary_by_user": self._build_timeclock_summary(entries if entries else active_entries, now=now),
            "active_entries": [self._serialize_timeclock_entry(entry, reference_now=now) for entry in active_entries],
        }

    def impersonate_tenant(self, session: SessionIdentity, tenant_id: str) -> SessionIdentity:
        if session.role != "super_admin":
            raise ValueError("Operazione riservata al super admin.")
        if tenant_id == SUPER_ADMIN_TENANT_ID:
            raise ValueError("Usa la sessione piattaforma per il pannello admin.")

        with self._lock:
            with self._connect_registry() as connection:
                tenant = connection.execute(
                    "SELECT id FROM tenants WHERE id = ? LIMIT 1",
                    (tenant_id,),
                ).fetchone()
                if tenant is None:
                    raise ValueError("Locale non trovato.")
                return self._create_session(connection, tenant_id=tenant_id, user_id=SUPER_ADMIN_USER_ID)

    def get_llm_settings(self, tenant_id: str, scope: str) -> tuple[dict[str, object], str | None]:
        with self._connect_registry() as connection:
            row = connection.execute(
                """
                SELECT config_json, updated_at
                FROM tenant_llm_settings
                WHERE tenant_id = ? AND scope = ?
                LIMIT 1
                """,
                (tenant_id, scope),
            ).fetchone()

        if row is None:
            return {}, None

        try:
            payload = json.loads(row["config_json"])
        except json.JSONDecodeError:
            return {}, row["updated_at"]

        if not isinstance(payload, dict):
            return {}, row["updated_at"]

        return payload, row["updated_at"]

    def upsert_llm_settings(self, tenant_id: str, scope: str, config: dict[str, object]) -> str:
        updated_at = _iso_now()
        serialized = json.dumps(config, ensure_ascii=True)

        with self._lock:
            with self._connect_registry() as connection:
                connection.execute(
                    """
                    INSERT INTO tenant_llm_settings (tenant_id, scope, config_json, updated_at)
                    VALUES (?, ?, ?, ?)
                    ON CONFLICT(tenant_id, scope) DO UPDATE SET
                        config_json = excluded.config_json,
                        updated_at = excluded.updated_at
                    """,
                    (tenant_id, scope, serialized, updated_at),
                )
                connection.commit()

        return updated_at

    def menu_assets_directory(self, tenant_id: str) -> Path:
        target = self._menu_assets_dir / tenant_id
        target.mkdir(parents=True, exist_ok=True)
        return target

    def fiscal_documents_directory(self, tenant_id: str) -> Path:
        target = self._fiscal_documents_dir / tenant_id
        target.mkdir(parents=True, exist_ok=True)
        return target

    def create_menu_asset(
        self,
        *,
        asset_id: str,
        tenant_id: str,
        original_name: str,
        display_name: str,
        mime_type: str,
        kind: str,
        file_size_bytes: int,
        storage_path: str,
        status: str,
        extracted_text: str = "",
        analysis_text: str = "",
        error_detail: str | None = None,
    ) -> MenuAssetRecord:
        timestamp = _iso_now()
        with self._lock:
            with self._connect_registry() as connection:
                connection.execute(
                    """
                    INSERT INTO tenant_menu_assets (
                        id,
                        tenant_id,
                        original_name,
                        display_name,
                        mime_type,
                        kind,
                        file_size_bytes,
                        storage_path,
                        extracted_text,
                        analysis_text,
                        status,
                        error_detail,
                        created_at,
                        updated_at
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    (
                        asset_id,
                        tenant_id,
                        original_name,
                        display_name,
                        mime_type,
                        kind,
                        file_size_bytes,
                        storage_path,
                        extracted_text,
                        analysis_text,
                        status,
                        error_detail,
                        timestamp,
                        timestamp,
                    ),
                )
                connection.commit()

        record = self.get_menu_asset(tenant_id, asset_id)
        if record is None:
            raise KeyError(f"Asset menu {asset_id} non trovato")
        return record

    def get_menu_asset(self, tenant_id: str, asset_id: str) -> MenuAssetRecord | None:
        with self._connect_registry() as connection:
            row = connection.execute(
                """
                SELECT *
                FROM tenant_menu_assets
                WHERE tenant_id = ? AND id = ?
                LIMIT 1
                """,
                (tenant_id, asset_id),
            ).fetchone()

        if row is None:
            return None

        return self._menu_asset_from_row(row)

    def list_menu_assets(self, tenant_id: str) -> list[MenuAssetRecord]:
        with self._connect_registry() as connection:
            rows = list(
                connection.execute(
                    """
                    SELECT *
                    FROM tenant_menu_assets
                    WHERE tenant_id = ?
                    ORDER BY created_at DESC, id DESC
                    """,
                    (tenant_id,),
                )
            )

        return [self._menu_asset_from_row(row) for row in rows]

    def update_menu_asset(
        self,
        tenant_id: str,
        asset_id: str,
        *,
        display_name: str | None = None,
        extracted_text: str | None = None,
        analysis_text: str | None = None,
        status: str | None = None,
        error_detail: str | None = None,
    ) -> MenuAssetRecord:
        assignments: list[str] = ["updated_at = ?"]
        values: list[object] = [_iso_now()]

        if display_name is not None:
            assignments.append("display_name = ?")
            values.append(display_name)

        if extracted_text is not None:
            assignments.append("extracted_text = ?")
            values.append(extracted_text)

        if analysis_text is not None:
            assignments.append("analysis_text = ?")
            values.append(analysis_text)

        if status is not None:
            assignments.append("status = ?")
            values.append(status)

        if error_detail is not None:
            assignments.append("error_detail = ?")
            values.append(error_detail)

        values.extend([tenant_id, asset_id])

        with self._lock:
            with self._connect_registry() as connection:
                connection.execute(
                    f"""
                    UPDATE tenant_menu_assets
                    SET {", ".join(assignments)}
                    WHERE tenant_id = ? AND id = ?
                    """,
                    values,
                )
                connection.commit()

        record = self.get_menu_asset(tenant_id, asset_id)
        if record is None:
            raise KeyError(f"Asset menu {asset_id} non trovato")
        return record

    def clear_menu_asset_error(self, tenant_id: str, asset_id: str) -> MenuAssetRecord:
        timestamp = _iso_now()
        with self._lock:
            with self._connect_registry() as connection:
                connection.execute(
                    """
                    UPDATE tenant_menu_assets
                    SET error_detail = NULL, updated_at = ?
                    WHERE tenant_id = ? AND id = ?
                    """,
                    (timestamp, tenant_id, asset_id),
                )
                connection.commit()

        record = self.get_menu_asset(tenant_id, asset_id)
        if record is None:
            raise KeyError(f"Asset menu {asset_id} non trovato")
        return record

    def delete_menu_asset(self, tenant_id: str, asset_id: str) -> MenuAssetRecord | None:
        existing = self.get_menu_asset(tenant_id, asset_id)
        if existing is None:
            return None

        with self._lock:
            with self._connect_registry() as connection:
                connection.execute(
                    "DELETE FROM tenant_menu_assets WHERE tenant_id = ? AND id = ?",
                    (tenant_id, asset_id),
                )
                connection.commit()

        return existing

    def create_fiscal_document(
        self,
        *,
        document_id: str,
        tenant_id: str,
        original_name: str,
        display_name: str,
        mime_type: str,
        kind: str,
        file_size_bytes: int,
        file_hash: str | None = None,
        storage_path: str,
        document_type: str,
        document_number: str | None = None,
        document_date: str | None = None,
        supplier_name: str | None = None,
        total_amount: float | None = None,
        currency: str = "EUR",
        summary_text: str = "",
        extracted_text: str = "",
        status: str = "processing",
        matching_status: str = "pending",
        review_status: str = "to_review",
        error_detail: str | None = None,
    ) -> FiscalDocumentRecord:
        timestamp = _iso_now()
        with self._lock:
            with self._connect_registry() as connection:
                connection.execute(
                    """
                    INSERT INTO tenant_fiscal_documents (
                        id,
                        tenant_id,
                        original_name,
                        display_name,
                        mime_type,
                        kind,
                        file_size_bytes,
                        file_hash,
                        storage_path,
                        document_type,
                        document_number,
                        document_date,
                        supplier_name,
                        total_amount,
                        currency,
                        summary_text,
                        extracted_text,
                        status,
                        matching_status,
                        review_status,
                        error_detail,
                        created_at,
                        updated_at
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    (
                        document_id,
                        tenant_id,
                        original_name,
                        display_name,
                        mime_type,
                        kind,
                        file_size_bytes,
                        file_hash,
                        storage_path,
                        document_type,
                        document_number,
                        document_date,
                        supplier_name,
                        total_amount,
                        currency,
                        summary_text,
                        extracted_text,
                        status,
                        matching_status,
                        review_status,
                        error_detail,
                        timestamp,
                        timestamp,
                    ),
                )
                connection.commit()

        record = self.get_fiscal_document(tenant_id, document_id)
        if record is None:
            raise KeyError(f"Documento fiscale {document_id} non trovato")
        return record

    def get_fiscal_document(self, tenant_id: str, document_id: str) -> FiscalDocumentRecord | None:
        with self._connect_registry() as connection:
            row = connection.execute(
                """
                SELECT *
                FROM tenant_fiscal_documents
                WHERE tenant_id = ? AND id = ?
                LIMIT 1
                """,
                (tenant_id, document_id),
            ).fetchone()

        if row is None:
            return None
        return self._fiscal_document_from_row(row)

    def get_fiscal_document_by_hash(self, tenant_id: str, file_hash: str) -> FiscalDocumentRecord | None:
        with self._connect_registry() as connection:
            row = connection.execute(
                """
                SELECT *
                FROM tenant_fiscal_documents
                WHERE tenant_id = ? AND file_hash = ?
                ORDER BY created_at DESC
                LIMIT 1
                """,
                (tenant_id, file_hash),
            ).fetchone()

        if row is None:
            return None
        return self._fiscal_document_from_row(row)

    def list_fiscal_documents(self, tenant_id: str) -> list[FiscalDocumentRecord]:
        with self._connect_registry() as connection:
            rows = list(
                connection.execute(
                    """
                    SELECT *
                    FROM tenant_fiscal_documents
                    WHERE tenant_id = ?
                    ORDER BY created_at DESC, id DESC
                    """,
                    (tenant_id,),
                )
            )
        records = [self._fiscal_document_from_row(row) for row in rows]
        deduped: list[FiscalDocumentRecord] = []
        seen_semantic_keys: set[tuple[object, ...]] = set()
        seen_hashes: set[str] = set()
        for record in records:
            semantic_key = self._fiscal_document_dedup_key(record)
            if semantic_key in seen_semantic_keys:
                continue
            file_hash = (record.file_hash or "").strip()
            if file_hash and file_hash in seen_hashes:
                continue
            seen_semantic_keys.add(semantic_key)
            if file_hash:
                seen_hashes.add(file_hash)
            deduped.append(record)
        return deduped

    def get_fiscal_document_settings(self, tenant_id: str) -> FiscalDocumentSettingsRecord:
        with self._connect_registry() as connection:
            row = connection.execute(
                """
                SELECT tenant_id, inbound_email, updated_at
                FROM tenant_fiscal_document_settings
                WHERE tenant_id = ?
                LIMIT 1
                """,
                (tenant_id,),
            ).fetchone()

        if row is None:
            return FiscalDocumentSettingsRecord(tenant_id=tenant_id, inbound_email=None, updated_at=None)

        return self._fiscal_document_settings_from_row(row)

    def upsert_fiscal_document_settings(self, tenant_id: str, *, inbound_email: str | None) -> FiscalDocumentSettingsRecord:
        updated_at = _iso_now()
        with self._lock:
            with self._connect_registry() as connection:
                connection.execute(
                    """
                    INSERT INTO tenant_fiscal_document_settings (tenant_id, inbound_email, updated_at)
                    VALUES (?, ?, ?)
                    ON CONFLICT(tenant_id) DO UPDATE SET
                        inbound_email = excluded.inbound_email,
                        updated_at = excluded.updated_at
                    """,
                    (tenant_id, inbound_email, updated_at),
                )
                connection.commit()

        return self.get_fiscal_document_settings(tenant_id)

    def update_fiscal_document(
        self,
        tenant_id: str,
        document_id: str,
        *,
        display_name: str | None = None,
        file_hash: str | None = None,
        document_type: str | None = None,
        document_number: str | None = None,
        document_date: str | None = None,
        supplier_name: str | None = None,
        total_amount: float | None = None,
        currency: str | None = None,
        summary_text: str | None = None,
        extracted_text: str | None = None,
        status: str | None = None,
        matching_status: str | None = None,
        review_status: str | None = None,
        error_detail: str | None = None,
    ) -> FiscalDocumentRecord:
        assignments: list[str] = ["updated_at = ?"]
        values: list[object] = [_iso_now()]

        for column, value in (
            ("display_name", display_name),
            ("file_hash", file_hash),
            ("document_type", document_type),
            ("document_number", document_number),
            ("document_date", document_date),
            ("supplier_name", supplier_name),
            ("total_amount", total_amount),
            ("currency", currency),
            ("summary_text", summary_text),
            ("extracted_text", extracted_text),
            ("status", status),
            ("matching_status", matching_status),
            ("review_status", review_status),
            ("error_detail", error_detail),
        ):
            if value is not None:
                assignments.append(f"{column} = ?")
                values.append(value)

        values.extend([tenant_id, document_id])
        with self._lock:
            with self._connect_registry() as connection:
                connection.execute(
                    f"""
                    UPDATE tenant_fiscal_documents
                    SET {", ".join(assignments)}
                    WHERE tenant_id = ? AND id = ?
                    """,
                    values,
                )
                connection.commit()

        record = self.get_fiscal_document(tenant_id, document_id)
        if record is None:
            raise KeyError(f"Documento fiscale {document_id} non trovato")
        return record

    def get_fiscal_document_inbox_item(
        self,
        tenant_id: str,
        *,
        message_id: str,
        attachment_id: str,
    ) -> FiscalDocumentInboxItemRecord | None:
        with self._connect_registry() as connection:
            row = connection.execute(
                """
                SELECT *
                FROM tenant_fiscal_document_inbox_items
                WHERE tenant_id = ? AND message_id = ? AND attachment_id = ?
                LIMIT 1
                """,
                (tenant_id, message_id, attachment_id),
            ).fetchone()

        if row is None:
            return None
        return self._fiscal_document_inbox_item_from_row(row)

    def list_fiscal_document_inbox_items(self, tenant_id: str, *, limit: int = 40) -> list[FiscalDocumentInboxItemRecord]:
        with self._connect_registry() as connection:
            rows = list(
                connection.execute(
                    """
                    SELECT *
                    FROM tenant_fiscal_document_inbox_items
                    WHERE tenant_id = ?
                    ORDER BY created_at DESC, id DESC
                    LIMIT ?
                    """,
                    (tenant_id, max(1, limit)),
                )
            )
        return [self._fiscal_document_inbox_item_from_row(row) for row in rows]

    def create_fiscal_document_inbox_item(
        self,
        *,
        item_id: str,
        tenant_id: str,
        message_id: str,
        attachment_id: str,
        file_hash: str | None = None,
        subject: str,
        sender: str | None,
        received_at: str | None,
        attachment_name: str,
        mime_type: str,
        sync_status: str,
        document_id: str | None = None,
        error_detail: str | None = None,
    ) -> FiscalDocumentInboxItemRecord:
        timestamp = _iso_now()
        with self._lock:
            with self._connect_registry() as connection:
                connection.execute(
                    """
                    INSERT INTO tenant_fiscal_document_inbox_items (
                        id,
                        tenant_id,
                        message_id,
                        attachment_id,
                        file_hash,
                        subject,
                        sender,
                        received_at,
                        attachment_name,
                        mime_type,
                        sync_status,
                        document_id,
                        error_detail,
                        created_at,
                        updated_at
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    (
                        item_id,
                        tenant_id,
                        message_id,
                        attachment_id,
                        file_hash,
                        subject,
                        sender,
                        received_at,
                        attachment_name,
                        mime_type,
                        sync_status,
                        document_id,
                        error_detail,
                        timestamp,
                        timestamp,
                    ),
                )
                connection.commit()

        record = self.get_fiscal_document_inbox_item(tenant_id, message_id=message_id, attachment_id=attachment_id)
        if record is None:
            raise KeyError(f"Elemento inbox fiscale {item_id} non trovato")
        return record

    def update_fiscal_document_inbox_item(
        self,
        tenant_id: str,
        *,
        message_id: str,
        attachment_id: str,
        subject: str | None = None,
        sender: str | None = None,
        received_at: str | None = None,
        attachment_name: str | None = None,
        mime_type: str | None = None,
        sync_status: str | None = None,
        document_id: str | None = None,
        error_detail: str | None = None,
    ) -> FiscalDocumentInboxItemRecord:
        assignments: list[str] = ["updated_at = ?"]
        values: list[object] = [_iso_now()]

        for column, value in (
            ("subject", subject),
            ("sender", sender),
            ("received_at", received_at),
            ("attachment_name", attachment_name),
            ("mime_type", mime_type),
            ("sync_status", sync_status),
            ("document_id", document_id),
            ("error_detail", error_detail),
        ):
            if value is not None:
                assignments.append(f"{column} = ?")
                values.append(value)

        values.extend([tenant_id, message_id, attachment_id])
        with self._lock:
            with self._connect_registry() as connection:
                connection.execute(
                    f"""
                    UPDATE tenant_fiscal_document_inbox_items
                    SET {", ".join(assignments)}
                    WHERE tenant_id = ? AND message_id = ? AND attachment_id = ?
                    """,
                    values,
                )
                connection.commit()

        record = self.get_fiscal_document_inbox_item(tenant_id, message_id=message_id, attachment_id=attachment_id)
        if record is None:
            raise KeyError(f"Elemento inbox fiscale {message_id}/{attachment_id} non trovato")
        return record

    def get_assistant_thread(self, thread_id: str) -> AssistantThreadRecord | None:
        with self._connect_registry() as connection:
            row = connection.execute(
                "SELECT * FROM assistant_threads WHERE id = ? LIMIT 1",
                (thread_id,),
            ).fetchone()
        if row is None:
            return None
        return self._assistant_thread_from_row(row)

    def get_active_assistant_thread(self, session: SessionIdentity, surface: str = "home") -> AssistantThreadRecord | None:
        with self._connect_registry() as connection:
            row = connection.execute(
                """
                SELECT *
                FROM assistant_threads
                WHERE tenant_id = ? AND user_id = ? AND surface = ? AND status = 'active'
                ORDER BY updated_at DESC, created_at DESC
                LIMIT 1
                """,
                (session.tenant_id, session.user_id, surface),
            ).fetchone()
        if row is None:
            return None
        return self._assistant_thread_from_row(row)

    def list_assistant_messages(self, thread_id: str, *, limit: int = 80) -> list[AssistantStoredMessage]:
        with self._connect_registry() as connection:
            rows = list(
                connection.execute(
                    """
                    SELECT *
                    FROM (
                        SELECT *
                        FROM assistant_messages
                        WHERE thread_id = ?
                        ORDER BY sort_index DESC
                        LIMIT ?
                    )
                    ORDER BY sort_index ASC
                    """,
                    (thread_id, limit),
                )
            )
        return [self._assistant_message_from_row(row) for row in rows]

    def create_assistant_thread(self, session: SessionIdentity, surface: str = "home") -> AssistantThreadRecord:
        thread_id = f"ath_{uuid.uuid4().hex}"
        timestamp = _iso_now()
        with self._lock:
            with self._connect_registry() as connection:
                connection.execute(
                    """
                    INSERT INTO assistant_threads (id, tenant_id, user_id, surface, status, state_json, created_at, updated_at)
                    VALUES (?, ?, ?, ?, 'active', ?, ?, ?)
                    """,
                    (thread_id, session.tenant_id, session.user_id, surface, "{}", timestamp, timestamp),
                )
                connection.commit()
        record = self.get_assistant_thread(thread_id)
        if record is None:
            raise KeyError(f"Thread assistente {thread_id} non creato")
        return record

    def ensure_assistant_thread(
        self,
        session: SessionIdentity,
        *,
        surface: str = "home",
        thread_id: str | None = None,
    ) -> AssistantThreadRecord:
        if thread_id:
            current = self.get_assistant_thread(thread_id)
            if (
                current is not None
                and current.status == "active"
                and current.tenant_id == session.tenant_id
                and current.user_id == session.user_id
                and current.surface == surface
            ):
                return current
        current = self.get_active_assistant_thread(session, surface)
        if current is not None:
            return current
        return self.create_assistant_thread(session, surface)

    def append_assistant_messages(
        self,
        thread_id: str,
        messages: list[dict[str, str]],
    ) -> tuple[AssistantThreadRecord, list[AssistantStoredMessage]]:
        sanitized = [
            {
                "role": str(message.get("role") or "").strip(),
                "content": str(message.get("content") or "").strip(),
            }
            for message in messages
            if str(message.get("role") or "").strip() in {"user", "assistant"} and str(message.get("content") or "").strip()
        ]
        if not sanitized:
            thread = self.get_assistant_thread(thread_id)
            if thread is None:
                raise KeyError(f"Thread assistente {thread_id} non trovato")
            return thread, []

        timestamp = _iso_now()
        created: list[AssistantStoredMessage] = []
        with self._lock:
            with self._connect_registry() as connection:
                thread_row = connection.execute(
                    "SELECT * FROM assistant_threads WHERE id = ? LIMIT 1",
                    (thread_id,),
                ).fetchone()
                if thread_row is None:
                    raise KeyError(f"Thread assistente {thread_id} non trovato")

                max_sort_index = connection.execute(
                    "SELECT COALESCE(MAX(sort_index), 0) FROM assistant_messages WHERE thread_id = ?",
                    (thread_id,),
                ).fetchone()[0]
                sort_index = int(max_sort_index or 0)

                for message in sanitized:
                    sort_index += 1
                    message_id = f"ams_{uuid.uuid4().hex}"
                    connection.execute(
                        """
                        INSERT INTO assistant_messages (id, thread_id, role, content, created_at, sort_index)
                        VALUES (?, ?, ?, ?, ?, ?)
                        """,
                        (message_id, thread_id, message["role"], message["content"], timestamp, sort_index),
                    )
                    created.append(
                        AssistantStoredMessage(
                            id=message_id,
                            thread_id=thread_id,
                            role=message["role"],
                            content=message["content"],
                            created_at=timestamp,
                            sort_index=sort_index,
                        )
                    )

                connection.execute(
                    """
                    UPDATE assistant_threads
                    SET updated_at = ?, status = 'active'
                    WHERE id = ?
                    """,
                    (timestamp, thread_id),
                )
                connection.commit()

        thread = self.get_assistant_thread(thread_id)
        if thread is None:
            raise KeyError(f"Thread assistente {thread_id} non trovato")
        return thread, created

    def archive_assistant_thread(
        self,
        session: SessionIdentity,
        *,
        surface: str = "home",
        thread_id: str | None = None,
    ) -> None:
        with self._lock:
            with self._connect_registry() as connection:
                if thread_id:
                    connection.execute(
                        """
                        UPDATE assistant_threads
                        SET status = 'archived', updated_at = ?
                        WHERE id = ? AND tenant_id = ? AND user_id = ? AND surface = ?
                        """,
                        (_iso_now(), thread_id, session.tenant_id, session.user_id, surface),
                    )
                else:
                    connection.execute(
                        """
                        UPDATE assistant_threads
                        SET status = 'archived', updated_at = ?
                        WHERE id IN (
                            SELECT id
                            FROM assistant_threads
                            WHERE tenant_id = ? AND user_id = ? AND surface = ? AND status = 'active'
                            ORDER BY updated_at DESC, created_at DESC
                            LIMIT 1
                        )
                        """,
                        (_iso_now(), session.tenant_id, session.user_id, surface),
                    )
                connection.commit()

    def get_assistant_thread_state(self, thread_id: str) -> dict[str, object]:
        thread = self.get_assistant_thread(thread_id)
        if thread is None or not thread.state_json:
            return {}
        try:
            payload = json.loads(thread.state_json)
        except json.JSONDecodeError:
            return {}
        return payload if isinstance(payload, dict) else {}

    def update_assistant_thread_state(self, thread_id: str, state: dict[str, object]) -> AssistantThreadRecord:
        serialized_state = json.dumps(state or {}, ensure_ascii=False, default=str)
        timestamp = _iso_now()
        with self._lock:
            with self._connect_registry() as connection:
                connection.execute(
                    """
                    UPDATE assistant_threads
                    SET state_json = ?, updated_at = ?
                    WHERE id = ?
                    """,
                    (serialized_state, timestamp, thread_id),
                )
                connection.commit()

        record = self.get_assistant_thread(thread_id)
        if record is None:
            raise KeyError(f"Thread assistente {thread_id} non trovato")
        return record

    def create_assistant_run(
        self,
        *,
        thread_id: str,
        session: SessionIdentity,
        surface: str,
        route: str,
        model: str,
        user_message: str,
        assistant_reply: str,
        trace: dict[str, object],
    ) -> AssistantRunRecord:
        run_id = f"arn_{uuid.uuid4().hex}"
        created_at = _iso_now()
        serialized_trace = json.dumps(trace, ensure_ascii=False, default=str)
        with self._lock:
            with self._connect_registry() as connection:
                connection.execute(
                    """
                    INSERT INTO assistant_runs (
                        id, thread_id, tenant_id, user_id, surface, route, model,
                        user_message, assistant_reply, trace_json, created_at
                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                    """,
                    (
                        run_id,
                        thread_id,
                        session.tenant_id,
                        session.user_id,
                        surface,
                        route,
                        model,
                        user_message,
                        assistant_reply,
                        serialized_trace,
                        created_at,
                    ),
                )
                connection.commit()
        with self._connect_registry() as connection:
            record = connection.execute("SELECT * FROM assistant_runs WHERE id = ? LIMIT 1", (run_id,)).fetchone()
        if record is None:
            raise KeyError(f"Run assistente {run_id} non trovato")
        return self._assistant_run_from_row(record)

    def list_assistant_runs(self, thread_id: str, *, limit: int = 20) -> list[AssistantRunRecord]:
        with self._connect_registry() as connection:
            rows = list(
                connection.execute(
                    """
                    SELECT *
                    FROM assistant_runs
                    WHERE thread_id = ?
                    ORDER BY created_at DESC
                    LIMIT ?
                    """,
                    (thread_id, limit),
                )
            )
        return [self._assistant_run_from_row(row) for row in rows]


_tenant_store: TenantStore | None = None


def get_tenant_store() -> TenantStore:
    global _tenant_store
    if _tenant_store is None:
        _tenant_store = TenantStore()
    return _tenant_store
