#!/usr/bin/env python3
"""ERPClaw Setup Skill — db_query.py

System administration actions for ERPClaw ERP.
Manages companies, currencies, payment terms, UoMs, seed data, and backups.

Usage: python3 db_query.py --action <action-name> [--flags ...]
Output: JSON to stdout, exit 0 on success, exit 1 on error.
"""
import argparse
import glob as glob_mod
import json
import os
import re
import shutil
import sqlite3
import sys
import urllib.error
import urllib.request
import uuid
from datetime import date, datetime, timedelta, timezone
from decimal import Decimal

# Add shared lib to path — installed location first, then bundled copy. The
# installed location resolves through the single ERPCLAW_HOME point of truth
# (ADR-0017): unset ⇒ ~/.openclaw/erpclaw/lib, byte-identical to the historical
# literal; Hermes sets ERPCLAW_HOME to its skill-dir install so the lib resolves
# under the Hermes prefix, not ~/.openclaw.
_LIB_INSTALLED = os.path.join(
    os.path.expanduser(os.environ.get("ERPCLAW_HOME", "~/.openclaw/erpclaw")), "lib")
_LIB_BUNDLED = os.path.join(os.path.dirname(os.path.abspath(__file__)), "lib")

_lib_found = False
for _lib_path in [_LIB_INSTALLED, _LIB_BUNDLED]:
    if os.path.isdir(os.path.join(_lib_path, "erpclaw_lib")):
        sys.path.insert(0, _lib_path)
        _lib_found = True
        break

if not _lib_found:
    print(json.dumps({
        "status": "error",
        "error": "erpclaw_lib not found. Re-install erpclaw: clawhub install erpclaw"
    }))
    sys.exit(1)

from erpclaw_lib.db import get_connection, ensure_db_exists, DEFAULT_DB_PATH
from erpclaw_lib.decimal_utils import to_decimal
from erpclaw_lib.validation import check_input_lengths
from erpclaw_lib.response import ok, err, row_to_dict
from erpclaw_lib.audit import audit
from erpclaw_lib.query import Q, P, Table, Field, fn, now, dynamic_update, insert_or_ignore
from erpclaw_lib.args import SafeArgumentParser, check_unknown_args
from erpclaw_lib import custom_fields as cf
from erpclaw_lib.vendor.pypika import Order
from erpclaw_lib.vendor.pypika.terms import LiteralValue


SKILL_DIR = os.path.dirname(os.path.abspath(__file__))
ASSETS_DIR = os.path.join(SKILL_DIR, "assets")
BACKUP_DIR = os.path.join(
    os.path.expanduser(os.environ.get("ERPCLAW_HOME", "~/.openclaw/erpclaw")), "backups")

INDUSTRY_PROFILE_MAP = {
    "retail": "retail", "store": "retail", "shop": "retail", "ecommerce": "retail",
    "restaurant": "food-service", "food": "food-service", "catering": "food-service", "cafe": "food-service",
    "healthcare": "healthcare", "medical": "healthcare", "clinic": "healthcare", "hospital": "healthcare",
    "dental": "dental", "dentist": "dental",
    "veterinary": "veterinary", "vet": "veterinary", "animal": "veterinary",
    "construction": "construction", "contractor": "construction", "builder": "construction",
    "manufacturing": "manufacturing", "factory": "manufacturing", "production": "manufacturing",
    "law": "legal", "legal": "legal", "attorney": "legal", "law-firm": "legal",
    "farm": "agriculture", "agriculture": "agriculture", "ranch": "agriculture", "farming": "agriculture",
    "hotel": "hospitality", "hospitality": "hospitality", "resort": "hospitality", "motel": "hospitality",
    "property": "property-management", "real-estate": "property-management", "landlord": "property-management",
    "school": "k12-school", "k12": "k12-school", "education": "k12-school",
    "university": "college-university", "college": "college-university", "higher-ed": "college-university",
    "nonprofit": "nonprofit", "charity": "nonprofit", "foundation": "nonprofit", "ngo": "nonprofit",
    "auto-repair": "automotive", "mechanic": "automotive", "dealership": "automotive", "automotive": "automotive",
    "therapy": "mental-health", "therapist": "mental-health", "counseling": "mental-health", "psychiatry": "mental-health",
    "home-health": "home-health", "home-care": "home-health",
    "consulting": "professional-services", "agency": "professional-services",
    "distribution": "distribution", "wholesale": "distribution",
    "saas": "saas", "subscription": "saas", "software": "saas",
}


# ---------------------------------------------------------------------------
# Company Management
# ---------------------------------------------------------------------------

def setup_company(conn, args):
    """Create a new company."""
    name = args.name
    if not name:
        err("--name is required")

    # Generate abbreviation from name initials
    abbr = args.abbr or "".join(w[0].upper() for w in name.split() if w)
    if not abbr:
        abbr = name[:3].upper()

    company_id = str(uuid.uuid4())
    try:
        t = Table("company")
        q = Q.into(t).columns(
            "id", "name", "abbr", "default_currency", "country",
            "fiscal_year_start_month",
        ).insert(P(), P(), P(), P(), P(), P())
        conn.execute(
            q.get_sql(),
            (company_id, name, abbr,
             args.currency or "USD",
             args.country or "United States",
             args.fiscal_year_start_month or 1),
        )
        audit(conn, "erpclaw-setup", "create", "company", company_id,
               new_values={"name": name, "abbr": abbr},
               description=f"Created company '{name}'")
        conn.commit()
    except sqlite3.IntegrityError as e:
        sys.stderr.write(f"[erpclaw-setup] {e}\n")
        err("Company creation failed — check for duplicates or invalid data")

    # Auto-create fiscal year for the company
    fy_id = str(uuid.uuid4())
    fiscal_month = int(args.fiscal_year_start_month or 1)
    today = date.today()
    fy_start_year = today.year if today.month >= fiscal_month else today.year - 1
    fy_start = date(fy_start_year, fiscal_month, 1)
    if fiscal_month == 1:
        fy_end = date(fy_start_year, 12, 31)
    else:
        fy_end = date(fy_start_year + 1, fiscal_month, 1) - timedelta(days=1)
    fy_name = f"{abbr} FY {fy_start.strftime('%Y-%m-%d')} to {fy_end.strftime('%Y-%m-%d')}"
    try:
        t_fy = Table("fiscal_year")
        q_fy = (Q.into(t_fy)
                .columns("id", "name", "start_date", "end_date", "company_id")
                .insert(P(), P(), P(), P(), P()))
        conn.execute(q_fy.get_sql(),
                     (fy_id, fy_name, fy_start.isoformat(), fy_end.isoformat(), company_id))
        conn.commit()
    except Exception as e:
        sys.stderr.write(f"[erpclaw-setup] Fiscal year creation failed: {e}\n")
        pass  # Non-fatal — user can create manually

    # Auto-create default cost center
    cc_id = str(uuid.uuid4())
    cc_name = f"{name} - Main"
    try:
        t_cc = Table("cost_center")
        q_cc = (Q.into(t_cc)
                .columns("id", "name", "company_id", "is_group")
                .insert(P(), P(), P(), P()))
        conn.execute(q_cc.get_sql(), (cc_id, cc_name, company_id, 0))
        # Set as company default
        t_co = Table("company")
        q_co = (Q.update(t_co).set(Field("default_cost_center_id"), P()).where(t_co.id == P()))
        conn.execute(q_co.get_sql(), (cc_id, company_id))
        conn.commit()
    except Exception as e:
        sys.stderr.write(f"[erpclaw-setup] Cost center creation failed: {e}\n")
        pass  # Non-fatal — user can create manually

    # Auto-create default warehouse
    wh_id = str(uuid.uuid4())
    wh_name = f"{name} - Main Warehouse"
    try:
        t_wh = Table("warehouse")
        q_wh = (Q.into(t_wh)
                .columns("id", "name", "company_id", "warehouse_type", "is_group")
                .insert(P(), P(), P(), P(), P()))
        conn.execute(q_wh.get_sql(), (wh_id, wh_name, company_id, "stores", 0))
        # Set as company default
        t_co2 = Table("company")
        q_co2 = (Q.update(t_co2).set(Field("default_warehouse_id"), P()).where(t_co2.id == P()))
        conn.execute(q_co2.get_sql(), (wh_id, company_id))
        conn.commit()
    except Exception as e:
        sys.stderr.write(f"[erpclaw-setup] Warehouse creation failed: {e}\n")
        pass  # Non-fatal — user can create manually

    # Auto-onboard if --industry is provided
    modules_installed = []
    modules_failed = []
    onboard_profile = None
    region_module = None
    if getattr(args, "industry", None):
        industry_key = args.industry.lower().replace(" ", "-")
        onboard_profile = INDUSTRY_PROFILE_MAP.get(industry_key)
        if onboard_profile:
            try:
                # Import onboarding module and call onboard programmatically
                sys.path.insert(0, os.path.join(SKILL_DIR, ".."))
                from onboarding import PROFILES, COUNTRY_REGION_MAP
                from module_manager import _load_registry, _install_module_inner, _registry_to_dict, build_action_cache

                profile = PROFILES.get(onboard_profile)
                if profile:
                    registry = _load_registry()
                    modules_by_name = _registry_to_dict(registry)
                    installed_rows = conn.execute(
                        "SELECT name FROM erpclaw_module WHERE install_status = 'installed'"
                    ).fetchall()
                    already_installed = {row["name"] for row in installed_rows}

                    for module_name in profile["modules"]:
                        if module_name in already_installed:
                            continue
                        if module_name not in modules_by_name:
                            continue
                        try:
                            install_args = argparse.Namespace(module_name=module_name)
                            _install_module_inner(install_args, conn, modules_by_name, depth=0)
                            modules_installed.append(module_name)
                            already_installed.add(module_name)
                        except SystemExit:
                            conn = get_connection()
                            check = conn.execute(
                                "SELECT install_status FROM erpclaw_module WHERE name = ?",
                                (module_name,)
                            ).fetchone()
                            if check and check["install_status"] == "installed":
                                modules_installed.append(module_name)
                                already_installed.add(module_name)
                        except Exception as e:
                            # Best-effort onboarding: one module failing must not
                            # abort the rest, but the failure must be visible —
                            # not silently swallowed (user asked for this module).
                            modules_failed.append({"module": module_name, "error": str(e)})
                            print(f"WARN: onboarding auto-install of module "
                                  f"'{module_name}' failed: {e}", file=sys.stderr)

                    # Also install regional module based on country
                    country_val = args.country or "United States"
                    region_module = COUNTRY_REGION_MAP.get(country_val)
                    if region_module and region_module not in already_installed:
                        if region_module in modules_by_name:
                            try:
                                install_args = argparse.Namespace(module_name=region_module)
                                _install_module_inner(install_args, conn, modules_by_name, depth=0)
                                modules_installed.append(region_module)
                            except SystemExit:
                                conn = get_connection()
                                check = conn.execute(
                                    "SELECT install_status FROM erpclaw_module WHERE name = ?",
                                    (region_module,)
                                ).fetchone()
                                if check and check["install_status"] == "installed":
                                    modules_installed.append(region_module)
                            except Exception as e:
                                modules_failed.append({"module": region_module, "error": str(e)})
                                print(f"WARN: onboarding auto-install of region module "
                                      f"'{region_module}' failed: {e}", file=sys.stderr)
            except ImportError:
                pass  # Onboarding not available — skip silently

    result = {"company_id": company_id, "name": name, "abbr": abbr,
              "fiscal_year_id": fy_id, "cost_center_id": cc_id, "warehouse_id": wh_id}
    if onboard_profile:
        result["onboard_profile"] = onboard_profile
    if modules_installed:
        result["modules_installed"] = modules_installed
    if modules_failed:
        result["modules_failed"] = modules_failed
    if region_module:
        result["region_module"] = region_module
    ok(result)


def update_company(conn, args):
    """Update company fields."""
    company_id = args.company_id
    if not company_id:
        # Default to first company
        t = Table("company")
        q = Q.from_(t).select(t.id).limit(1)
        row = conn.execute(q.get_sql()).fetchone()
        if not row:
            err("No company found",
                 suggestion="Run 'tutorial' to create a demo company, or 'setup company' to create your own.")
        company_id = row["id"]

    t = Table("company")
    q = Q.from_(t).select(t.star).where(t.id == P())
    old = row_to_dict(conn.execute(q.get_sql(), (company_id,)).fetchone())
    if not old:
        err(f"Company {company_id} not found")

    updatable = [
        "name", "abbr", "default_currency", "country", "tax_id",
        "default_receivable_account_id", "default_payable_account_id",
        "default_income_account_id", "default_expense_account_id",
        "default_cost_center_id", "default_warehouse_id",
        "default_bank_account_id", "default_cash_account_id",
        "round_off_account_id", "exchange_gain_loss_account_id",
        "perpetual_inventory", "enable_negative_stock",
        "accounts_frozen_till_date", "role_allowed_for_frozen_entries",
        "fiscal_year_start_month",
    ]

    updates = {}
    for field in updatable:
        val = getattr(args, field.replace("-", "_"), None)
        if val is not None:
            updates[field] = val

    if not updates:
        err("No fields to update. Use --field-name value flags.")

    # Validate that default account IDs are not group accounts
    account_fields = [
        "default_receivable_account_id", "default_payable_account_id",
        "default_income_account_id", "default_expense_account_id",
        "default_bank_account_id", "default_cash_account_id",
        "round_off_account_id", "exchange_gain_loss_account_id",
    ]
    for field in account_fields:
        if field in updates and updates[field]:
            ta = Table("account")
            qa = Q.from_(ta).select(ta.name, ta.account_number, ta.is_group).where(ta.id == P())
            acct = conn.execute(qa.get_sql(), (updates[field],)).fetchone()
            if acct and acct["is_group"]:
                err(
                    f"Cannot set {field} to group account "
                    f"'{acct['name']}' ({acct['account_number']}). "
                    f"Use a leaf account (is_group=0) instead."
                )

    tc = Table("company")
    qu = Q.update(tc)
    for k in updates:
        qu = qu.set(Field(k), P())
    qu = qu.set(Field("updated_at"), now())
    qu = qu.where(tc.id == P())
    values = list(updates.values()) + [company_id]
    conn.execute(qu.get_sql(), values)
    audit(conn, "erpclaw-setup", "update", "company", company_id,
           old_values={k: old.get(k) for k in updates},
           new_values=updates,
           description=f"Updated company fields: {list(updates.keys())}")
    conn.commit()

    ok({"company_id": company_id, "updated_fields": list(updates.keys())})


def get_company(conn, args):
    """Get a single company record."""
    company_id = args.company_id
    t = Table("company")
    if not company_id:
        q = Q.from_(t).select(t.star).limit(1)
        row = conn.execute(q.get_sql()).fetchone()
    else:
        q = Q.from_(t).select(t.star).where(t.id == P())
        row = conn.execute(q.get_sql(), (company_id,)).fetchone()
    if not row:
        err("No company found",
             suggestion="Run 'tutorial' to create a demo company, or 'setup company' to create your own.")
    ok({"company": row_to_dict(row)})


def list_companies(conn, args):
    """List all companies."""
    limit = int(args.limit or 20)
    offset = int(args.offset or 0)
    t = Table("company")
    qc = Q.from_(t).select(fn.Count("*").as_("cnt"))
    total_count = conn.execute(qc.get_sql()).fetchone()["cnt"]
    q = Q.from_(t).select(t.star).orderby(t.name).limit(limit).offset(offset)
    rows = conn.execute(q.get_sql()).fetchall()
    ok({"companies": [row_to_dict(r) for r in rows],
         "total_count": total_count, "limit": limit, "offset": offset,
         "has_more": offset + limit < total_count})


# ---------------------------------------------------------------------------
# Currency Management
# ---------------------------------------------------------------------------

def add_currency(conn, args):
    """Add a currency."""
    code = args.code
    if not code:
        err("--code is required")
    name = args.name or code
    try:
        t = Table("currency")
        q = Q.into(t).columns("code", "name", "symbol", "decimal_places", "enabled").insert(
            P(), P(), P(), P(), P()
        )
        conn.execute(
            q.get_sql(),
            (code.upper(), name, args.symbol or "", args.decimal_places or 2,
             1 if args.enabled else 0),
        )
        audit(conn, "erpclaw-setup", "create", "currency", code,
               new_values={"code": code, "name": name})
        conn.commit()
    except sqlite3.IntegrityError:
        err(f"Currency '{code}' already exists")
    ok({"code": code.upper(), "name": name})


def list_currencies(conn, args):
    """List currencies, optionally only enabled."""
    limit = int(args.limit or 20)
    offset = int(args.offset or 0)
    t = Table("currency")
    if args.enabled_only:
        qc = Q.from_(t).select(fn.Count("*").as_("cnt")).where(t.enabled == 1)
        total_count = conn.execute(qc.get_sql()).fetchone()["cnt"]
        q = Q.from_(t).select(t.star).where(t.enabled == 1).orderby(t.code).limit(limit).offset(offset)
        rows = conn.execute(q.get_sql()).fetchall()
    else:
        qc = Q.from_(t).select(fn.Count("*").as_("cnt"))
        total_count = conn.execute(qc.get_sql()).fetchone()["cnt"]
        q = Q.from_(t).select(t.star).orderby(t.code).limit(limit).offset(offset)
        rows = conn.execute(q.get_sql()).fetchall()
    ok({"currencies": [row_to_dict(r) for r in rows],
         "total_count": total_count, "limit": limit, "offset": offset,
         "has_more": offset + limit < total_count})


def add_exchange_rate(conn, args):
    """Add an exchange rate."""
    if not args.from_currency or not args.to_currency or not args.rate:
        err("--from-currency, --to-currency, and --rate are required")

    rate_id = str(uuid.uuid4())
    effective_date = args.effective_date or datetime.now(timezone.utc).strftime("%Y-%m-%d")
    source = args.source or "manual"

    try:
        t = Table("exchange_rate")
        q = Q.into(t).columns(
            "id", "from_currency", "to_currency", "rate", "effective_date", "source"
        ).insert(P(), P(), P(), P(), P(), P())
        conn.execute(
            q.get_sql(),
            (rate_id, args.from_currency.upper(), args.to_currency.upper(),
             args.rate, effective_date, source),
        )
        audit(conn, "erpclaw-setup", "create", "exchange_rate", rate_id,
               new_values={"from": args.from_currency, "to": args.to_currency,
                           "rate": args.rate, "date": effective_date})
        conn.commit()
    except sqlite3.IntegrityError as e:
        sys.stderr.write(f"[erpclaw-setup] {e}\n")
        err("Exchange rate creation failed — check for duplicates or invalid data")

    ok({"exchange_rate_id": rate_id, "effective_date": effective_date})


def get_exchange_rate(conn, args):
    """Get exchange rate for a currency pair on a date (or most recent before)."""
    if not args.from_currency or not args.to_currency:
        err("--from-currency and --to-currency are required")

    date = args.effective_date or datetime.now(timezone.utc).strftime("%Y-%m-%d")
    t = Table("exchange_rate")
    q = (Q.from_(t).select(t.star)
         .where(t.from_currency == P())
         .where(t.to_currency == P())
         .where(t.effective_date <= P())
         .orderby(t.effective_date, order=Order.desc)
         .limit(1))
    row = conn.execute(
        q.get_sql(),
        (args.from_currency.upper(), args.to_currency.upper(), date),
    ).fetchone()

    if not row:
        err(f"No exchange rate found for {args.from_currency}/{args.to_currency} on or before {date}")
    ok({"rate": row["rate"], "effective_date": row["effective_date"],
         "source": row["source"]})


def list_exchange_rates(conn, args):
    """List exchange rates with optional filters."""
    limit = int(args.limit or 20)
    offset = int(args.offset or 0)
    t = Table("exchange_rate")
    params = []
    qc = Q.from_(t).select(fn.Count("*").as_("cnt"))
    q = Q.from_(t).select(t.star)
    if args.from_currency:
        qc = qc.where(t.from_currency == P())
        q = q.where(t.from_currency == P())
        params.append(args.from_currency.upper())
    if args.to_currency:
        qc = qc.where(t.to_currency == P())
        q = q.where(t.to_currency == P())
        params.append(args.to_currency.upper())
    if args.from_date:
        qc = qc.where(t.effective_date >= P())
        q = q.where(t.effective_date >= P())
        params.append(args.from_date)
    if args.to_date:
        qc = qc.where(t.effective_date <= P())
        q = q.where(t.effective_date <= P())
        params.append(args.to_date)

    total_count = conn.execute(qc.get_sql(), params).fetchone()["cnt"]

    q = q.orderby(t.effective_date, order=Order.desc).limit(limit).offset(offset)
    rows = conn.execute(q.get_sql(), params + []).fetchall()
    ok({"rates": [row_to_dict(r) for r in rows],
         "total_count": total_count, "limit": limit, "offset": offset,
         "has_more": offset + limit < total_count})


# ---------------------------------------------------------------------------
# Payment Terms
# ---------------------------------------------------------------------------

def add_payment_terms(conn, args):
    """Add payment terms."""
    name = args.name
    if not name:
        err("--name is required")

    pt_id = str(uuid.uuid4())
    try:
        t = Table("payment_terms")
        q = Q.into(t).columns(
            "id", "name", "due_days", "discount_percentage", "discount_days", "description"
        ).insert(P(), P(), P(), P(), P(), P())
        conn.execute(
            q.get_sql(),
            (pt_id, name, args.due_days or 30,
             args.discount_percentage, args.discount_days,
             args.description),
        )
        audit(conn, "erpclaw-setup", "create", "payment_terms", pt_id,
               new_values={"name": name})
        conn.commit()
    except sqlite3.IntegrityError:
        err(f"Payment terms '{name}' already exists")
    ok({"payment_terms_id": pt_id, "name": name})


def list_payment_terms(conn, args):
    """List all payment terms."""
    limit = int(args.limit or 20)
    offset = int(args.offset or 0)
    t = Table("payment_terms")
    qc = Q.from_(t).select(fn.Count("*").as_("cnt"))
    total_count = conn.execute(qc.get_sql()).fetchone()["cnt"]
    q = Q.from_(t).select(t.star).orderby(t.due_days).orderby(t.name).limit(limit).offset(offset)
    rows = conn.execute(q.get_sql()).fetchall()
    ok({"terms": [row_to_dict(r) for r in rows],
         "total_count": total_count, "limit": limit, "offset": offset,
         "has_more": offset + limit < total_count})


# ---------------------------------------------------------------------------
# Units of Measure
# ---------------------------------------------------------------------------

def add_uom(conn, args):
    """Add a unit of measure."""
    name = args.name
    if not name:
        err("--name is required")

    uom_id = str(uuid.uuid4())
    try:
        t = Table("uom")
        q = Q.into(t).columns("id", "name", "must_be_whole_number").insert(P(), P(), P())
        conn.execute(
            q.get_sql(),
            (uom_id, name, 1 if args.must_be_whole_number else 0),
        )
        audit(conn, "erpclaw-setup", "create", "uom", uom_id, new_values={"name": name})
        conn.commit()
    except sqlite3.IntegrityError:
        err(f"UoM '{name}' already exists")
    ok({"uom_id": uom_id, "name": name})


def list_uoms(conn, args):
    """List all units of measure."""
    limit = int(args.limit or 20)
    offset = int(args.offset or 0)
    t = Table("uom")
    qc = Q.from_(t).select(fn.Count("*").as_("cnt"))
    total_count = conn.execute(qc.get_sql()).fetchone()["cnt"]
    q = Q.from_(t).select(t.star).orderby(t.name).limit(limit).offset(offset)
    rows = conn.execute(q.get_sql()).fetchall()
    ok({"uoms": [row_to_dict(r) for r in rows],
         "total_count": total_count, "limit": limit, "offset": offset,
         "has_more": offset + limit < total_count})


def add_uom_conversion(conn, args):
    """Add a UoM conversion factor."""
    if not args.from_uom or not args.to_uom or not args.conversion_factor:
        err("--from-uom, --to-uom, and --conversion-factor are required")

    conv_id = str(uuid.uuid4())
    try:
        t = Table("uom_conversion")
        q = Q.into(t).columns(
            "id", "from_uom", "to_uom", "conversion_factor", "item_id"
        ).insert(P(), P(), P(), P(), P())
        conn.execute(
            q.get_sql(),
            (conv_id, args.from_uom, args.to_uom,
             args.conversion_factor, args.item_id),
        )
        audit(conn, "erpclaw-setup", "create", "uom_conversion", conv_id,
               new_values={"from": args.from_uom, "to": args.to_uom,
                           "factor": args.conversion_factor})
        conn.commit()
    except sqlite3.IntegrityError as e:
        sys.stderr.write(f"[erpclaw-setup] {e}\n")
        err("UoM conversion creation failed — check for duplicates or invalid data")
    ok({"uom_conversion_id": conv_id})


# ---------------------------------------------------------------------------
# System Operations
# ---------------------------------------------------------------------------

def seed_defaults(conn, args):
    """Load standard seed data (currencies, UoMs, payment terms). Idempotent."""
    company_id = args.company_id
    if not company_id:
        t = Table("company")
        q = Q.from_(t).select(t.id).limit(1)
        row = conn.execute(q.get_sql()).fetchone()
        if not row:
            err("No company found. Run setup-company first.",
                 suggestion="Run 'tutorial' to create a demo company, or 'setup company' to create your own.")
        company_id = row["id"]

    counts = {"currencies_seeded": 0, "uoms_seeded": 0, "payment_terms_seeded": 0}

    # Seed currencies
    currencies_file = os.path.join(ASSETS_DIR, "currencies.json")
    if os.path.exists(currencies_file):
        with open(currencies_file) as f:
            currencies = json.load(f)
        for c in currencies:
            try:
                conn.execute(
                    insert_or_ignore("INSERT OR IGNORE INTO currency (code, name, symbol, decimal_places, enabled) VALUES (?, ?, ?, ?, ?)"),
                    (c["code"], c["name"], c.get("symbol", ""),
                     c.get("decimal_places", 2), c.get("enabled", 0)),
                )
                counts["currencies_seeded"] += 1
            except sqlite3.IntegrityError:
                pass

    # Seed UoMs
    uom_file = os.path.join(ASSETS_DIR, "default_uom.json")
    if os.path.exists(uom_file):
        with open(uom_file) as f:
            uoms = json.load(f)
        for u in uoms:
            try:
                conn.execute(
                    insert_or_ignore("INSERT OR IGNORE INTO uom (id, name, must_be_whole_number) VALUES (?, ?, ?)"),
                    (str(uuid.uuid4()), u["name"], u.get("must_be_whole_number", 0)),
                )
                counts["uoms_seeded"] += 1
            except sqlite3.IntegrityError:
                pass

    # Seed payment terms
    pt_file = os.path.join(ASSETS_DIR, "default_payment_terms.json")
    if os.path.exists(pt_file):
        with open(pt_file) as f:
            terms = json.load(f)
        for t in terms:
            try:
                conn.execute(
                    insert_or_ignore("""INSERT OR IGNORE INTO payment_terms
                       (id, name, due_days, discount_percentage, discount_days, description)
                       VALUES (?, ?, ?, ?, ?, ?)"""),
                    (str(uuid.uuid4()), t["name"], t.get("due_days", 30),
                     t.get("discount_percentage"), t.get("discount_days"),
                     t.get("description")),
                )
                counts["payment_terms_seeded"] += 1
            except sqlite3.IntegrityError:
                pass

    audit(conn, "erpclaw-setup", "seed", "system", company_id,
           new_values=counts, description="Seeded default data")
    conn.commit()
    ok(counts)


def get_audit_log(conn, args):
    """Query audit log with optional filters."""
    t = Table("audit_log")
    q = Q.from_(t).select(t.star)
    params = []
    if args.entity_type:
        q = q.where(t.entity_type == P())
        params.append(args.entity_type)
    if args.entity_id:
        q = q.where(t.entity_id == P())
        params.append(args.entity_id)
    if args.audit_action:
        q = q.where(t.action == P())
        params.append(args.audit_action)
    if args.from_date:
        q = q.where(t.timestamp >= P())
        params.append(args.from_date)
    if args.to_date:
        q = q.where(t.timestamp <= P())
        params.append(args.to_date)
    limit = int(args.limit or 50)
    q = q.orderby(t.timestamp, order=Order.desc).limit(limit)

    rows = conn.execute(q.get_sql(), params).fetchall()
    entries = []
    for r in rows:
        entry = row_to_dict(r)
        # Parse JSON fields
        if entry.get("old_values"):
            entry["old_values"] = json.loads(entry["old_values"])
        if entry.get("new_values"):
            entry["new_values"] = json.loads(entry["new_values"])
        entries.append(entry)
    ok({"entries": entries})


def get_schema_version(conn, args):
    """Read schema version for a module."""
    module = args.module or "erpclaw-setup"
    t = Table("schema_version")
    q = Q.from_(t).select(t.star).where(t.module == P())
    row = conn.execute(q.get_sql(), (module,)).fetchone()
    if not row:
        err(f"No schema version found for module '{module}'")
    ok({"module": row["module"], "version": row["version"],
         "updated_at": row["updated_at"]})


def update_regional_settings(conn, args):
    """Set company-level regional settings."""
    company_id = args.company_id
    if not company_id:
        t = Table("company")
        q = Q.from_(t).select(t.id).limit(1)
        row = conn.execute(q.get_sql()).fetchone()
        if not row:
            err("No company found",
                 suggestion="Run 'tutorial' to create a demo company, or 'setup company' to create your own.")
        company_id = row["id"]

    settings = {}
    if args.date_format:
        settings["date_format"] = args.date_format
    if args.number_format:
        settings["number_format"] = args.number_format
    if args.default_tax_template_id:
        settings["default_tax_template_id"] = args.default_tax_template_id

    if not settings:
        err("No settings to update. Use --date-format, --number-format, etc.")

    for key, value in settings.items():
        conn.execute(
            """INSERT INTO regional_settings (id, company_id, key, value)
               VALUES (?, ?, ?, ?)
               ON CONFLICT(company_id, key) DO UPDATE SET value = ?, updated_at = CAST(CURRENT_TIMESTAMP AS TEXT)""",
            (str(uuid.uuid4()), company_id, key, value, value),
        )

    audit(conn, "erpclaw-setup", "update", "regional_settings", company_id,
           new_values=settings)
    conn.commit()
    ok({"updated": list(settings.keys())})


def backup_database(conn, args):
    """Create a backup of the database.

    Supports optional encryption with --encrypt --passphrase flags.
    Encrypted backups use AES-256 + HMAC-SHA256 authentication.
    """
    db_path = args.db_path or DEFAULT_DB_PATH
    backup_path = args.backup_path
    encrypt = getattr(args, "encrypt", False)
    passphrase = getattr(args, "passphrase", None)

    if encrypt and not passphrase:
        err("--passphrase is required when using --encrypt")

    if not backup_path:
        os.makedirs(BACKUP_DIR, exist_ok=True)
        ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
        ext = ".sqlite.enc" if encrypt else ".sqlite"
        backup_path = os.path.join(BACKUP_DIR, f"erpclaw_backup_{ts}{ext}")

    os.makedirs(os.path.dirname(backup_path), exist_ok=True)

    if encrypt:
        # First create unencrypted backup to temp file, then encrypt
        import tempfile
        with tempfile.NamedTemporaryFile(suffix=".sqlite", delete=False) as tmp:
            tmp_path = tmp.name
        try:
            src = sqlite3.connect(db_path)
            dst = sqlite3.connect(tmp_path)
            src.backup(dst)
            dst.close()
            src.close()

            from erpclaw_lib.crypto import encrypt_file, wrap_master_key
            from erpclaw_lib.master_key import master_key_exists, get_or_create_master_key

            # If a column-encryption master key exists on this machine,
            # wrap it with the backup passphrase and embed in the backup
            # header. This enables cross-machine restore (the receiving
            # machine can extract + install the master key via
            # import-master-key-from-backup).
            wrapped = None
            if master_key_exists():
                wrapped = wrap_master_key(get_or_create_master_key(), passphrase)

            enc_result = encrypt_file(tmp_path, backup_path, passphrase,
                                       wrapped_master_key=wrapped)
        finally:
            if os.path.exists(tmp_path):
                os.unlink(tmp_path)

        _chmod_600(backup_path)
        size = os.path.getsize(backup_path)
        ok({"backup_path": backup_path, "size_bytes": size,
             "encrypted": True,
             "carries_master_key": wrapped is not None,
             "original_size": enc_result["original_size"],
             "timestamp": datetime.now(timezone.utc).isoformat()})
    else:
        # Standard unencrypted backup
        src = sqlite3.connect(db_path)
        dst = sqlite3.connect(backup_path)
        src.backup(dst)
        dst.close()
        src.close()

        _chmod_600(backup_path)
        size = os.path.getsize(backup_path)
        ok({"backup_path": backup_path, "size_bytes": size,
             "encrypted": False,
             "timestamp": datetime.now(timezone.utc).isoformat()})


def list_backups(conn, args):
    """List all backup files with size, date, and validity."""
    backup_dir = BACKUP_DIR
    if not os.path.exists(backup_dir):
        ok({"backups": [], "count": 0, "total_size_bytes": 0})

    # Match both plain (.sqlite) and encrypted (.enc) backups
    sqlite_files = glob_mod.glob(os.path.join(backup_dir, "erpclaw_*.sqlite"))
    enc_files = glob_mod.glob(os.path.join(backup_dir, "erpclaw_*.enc"))
    files = sorted(sqlite_files + enc_files, reverse=True)

    backups = []
    total_size = 0
    for f in files:
        size = os.path.getsize(f)
        total_size += size
        # Parse timestamp from filename
        basename = os.path.basename(f)
        encrypted = basename.endswith(".enc")
        timestamp = None
        try:
            parts = basename.split("_", 2)
            if len(parts) >= 3:
                ts_str = parts[2].replace(".sqlite", "").replace(".enc", "")
                timestamp = datetime.strptime(ts_str, "%Y%m%d_%H%M%S").isoformat()
        except (ValueError, IndexError):
            pass

        backups.append({
            "path": f,
            "filename": basename,
            "size_bytes": size,
            "timestamp": timestamp,
            "encrypted": encrypted,
        })

    ok({"backups": backups, "count": len(backups),
         "total_size_bytes": total_size})


def verify_backup(conn, args):
    """Verify a backup file is a valid ERPClaw database without restoring.

    Auto-detects encrypted backups. Requires --passphrase for encrypted files.

    Checks:
    1. File exists and is valid SQLite (or encrypted backup)
    2. Contains schema_version table (ERPClaw signature)
    3. Passes PRAGMA integrity_check
    4. Reports table count and schema version
    """
    backup_path = args.backup_path
    passphrase = getattr(args, "passphrase", None)
    if not backup_path:
        err("--backup-path is required")
    if not os.path.exists(backup_path):
        err(f"File not found: {backup_path}")

    from erpclaw_lib.crypto import is_encrypted_backup, decrypt_file as crypto_decrypt
    encrypted = is_encrypted_backup(backup_path)
    verify_path = backup_path
    decrypted_tmp = None

    if encrypted:
        if not passphrase:
            ok({"encrypted": True, "valid": None,
                 "message": "Encrypted backup — provide --passphrase to verify contents"})
        import tempfile
        decrypted_tmp = tempfile.mktemp(suffix=".sqlite")
        try:
            crypto_decrypt(backup_path, decrypted_tmp, passphrase)
        except ValueError as e:
            if os.path.exists(decrypted_tmp):
                os.unlink(decrypted_tmp)
            err(str(e))
        verify_path = decrypted_tmp

    try:
        test_conn = sqlite3.connect(verify_path)
        test_conn.row_factory = sqlite3.Row

        tables = test_conn.execute(
            "SELECT COUNT(*) as cnt FROM sqlite_master WHERE type='table'"
        ).fetchone()["cnt"]

        has_schema = test_conn.execute(
            "SELECT COUNT(*) as cnt FROM sqlite_master "
            "WHERE type='table' AND name='schema_version'"
        ).fetchone()["cnt"]
        if has_schema == 0:
            test_conn.close()
            err("Not a valid ERPClaw database (missing schema_version table)")

        sv = test_conn.execute(
            "SELECT module, version FROM schema_version ORDER BY updated_at DESC LIMIT 1"
        ).fetchone()

        integrity = test_conn.execute("PRAGMA integrity_check").fetchone()[0]

        size = os.path.getsize(backup_path)
        test_conn.close()

        ok({
            "valid": integrity == "ok",
            "integrity": integrity,
            "tables": tables,
            "schema_module": sv["module"] if sv else None,
            "schema_version": sv["version"] if sv else None,
            "size_bytes": size,
            "encrypted": encrypted,
        })

    except sqlite3.DatabaseError as e:
        err(f"Not a valid SQLite file: {e}")
    finally:
        if decrypted_tmp and os.path.exists(decrypted_tmp):
            os.unlink(decrypted_tmp)


def restore_database(conn, args):
    """Restore the database from a backup file.

    Auto-detects encrypted backups. Requires --passphrase for encrypted files.

    Steps:
    1. Detect encrypted backup, decrypt if needed
    2. Validate backup is valid SQLite + ERPClaw schema
    3. Create safety backup of current DB
    4. Copy backup over current DB
    5. Verify integrity
    If any step fails, rollback to safety backup.
    """
    backup_path = args.backup_path
    passphrase = getattr(args, "passphrase", None)
    if not backup_path:
        err("--backup-path is required")
    if not os.path.exists(backup_path):
        err(f"Backup file not found: {backup_path}")

    # Step 0: Auto-detect encrypted backup
    from erpclaw_lib.crypto import is_encrypted_backup, decrypt_file as crypto_decrypt
    encrypted = is_encrypted_backup(backup_path)
    decrypted_tmp = None

    if encrypted:
        if not passphrase:
            err("Backup is encrypted — --passphrase is required",
                 suggestion="Use --passphrase to provide the encryption passphrase")
        import tempfile
        decrypted_tmp = tempfile.mktemp(suffix=".sqlite")
        try:
            crypto_decrypt(backup_path, decrypted_tmp, passphrase)
        except ValueError as e:
            if os.path.exists(decrypted_tmp):
                os.unlink(decrypted_tmp)
            err(str(e))
        restore_source = decrypted_tmp
    else:
        restore_source = backup_path

    # Step 1: Validate backup is a valid SQLite database
    try:
        test_conn = sqlite3.connect(restore_source)
        test_conn.execute("SELECT COUNT(*) FROM sqlite_master WHERE type='table'")
        sv = test_conn.execute(
            "SELECT COUNT(*) as cnt FROM sqlite_master WHERE type='table' AND name='schema_version'"
        ).fetchone()[0]
        if sv == 0:
            test_conn.close()
            err("Backup is not a valid ERPClaw database (missing schema_version table)")
        test_conn.close()
    except sqlite3.DatabaseError as e:
        if decrypted_tmp and os.path.exists(decrypted_tmp):
            os.unlink(decrypted_tmp)
        err(f"Backup is not a valid SQLite file: {e}")

    # Determine target DB path
    db_path = args.db_path or DEFAULT_DB_PATH

    # Step 2: Create safety backup of current DB
    os.makedirs(BACKUP_DIR, exist_ok=True)
    ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
    safety_path = os.path.join(BACKUP_DIR, f"erpclaw_pre_restore_{ts}.sqlite")

    try:
        # Close the passed-in connection so we can safely copy the file
        conn.close()

        if os.path.exists(db_path):
            shutil.copy2(db_path, safety_path)

        # Step 3: Copy backup over current DB
        shutil.copy2(restore_source, db_path)
        chmod_db_files(db_path)

        # Step 4 & 5: Verify restored DB
        verify_conn = sqlite3.connect(db_path)
        verify_conn.row_factory = sqlite3.Row

        # Integrity check
        result = verify_conn.execute("PRAGMA integrity_check").fetchone()
        if result[0] != "ok":
            verify_conn.close()
            raise ValueError(f"Integrity check failed: {result[0]}")

        # Verify schema_version exists
        sv_count = verify_conn.execute(
            "SELECT COUNT(*) as cnt FROM schema_version"
        ).fetchone()["cnt"]

        size = os.path.getsize(db_path)
        verify_conn.close()

        # Clean up decrypted temp file
        if decrypted_tmp and os.path.exists(decrypted_tmp):
            os.unlink(decrypted_tmp)

        ok({
            "restored_from": backup_path,
            "safety_backup": safety_path,
            "size_bytes": size,
            "schema_versions": sv_count,
            "integrity": "ok",
            "was_encrypted": encrypted,
        })

    except (ValueError, sqlite3.Error, OSError) as e:
        # Clean up decrypted temp file
        if decrypted_tmp and os.path.exists(decrypted_tmp):
            os.unlink(decrypted_tmp)
        # Rollback: restore safety backup
        if os.path.exists(safety_path):
            shutil.copy2(safety_path, db_path)
        err(f"Restore failed (rolled back to previous state): {e}")


def status(conn, args):
    """Overall system status."""
    tc = Table("company")
    tcu = Table("currency")
    tu = Table("uom")
    tpt = Table("payment_terms")
    tsv = Table("schema_version")
    companies = conn.execute(
        Q.from_(tc).select(fn.Count("*").as_("cnt")).get_sql()
    ).fetchone()["cnt"]
    currencies = conn.execute(
        Q.from_(tcu).select(fn.Count("*").as_("cnt")).where(tcu.enabled == 1).get_sql()
    ).fetchone()["cnt"]
    uoms = conn.execute(
        Q.from_(tu).select(fn.Count("*").as_("cnt")).get_sql()
    ).fetchone()["cnt"]
    payment_terms = conn.execute(
        Q.from_(tpt).select(fn.Count("*").as_("cnt")).get_sql()
    ).fetchone()["cnt"]

    versions = {}
    q = Q.from_(tsv).select(tsv.module, tsv.version)
    for row in conn.execute(q.get_sql()).fetchall():
        versions[row["module"]] = row["version"]

    ok({
        "companies": companies,
        "currencies": currencies,
        "uoms": uoms,
        "payment_terms": payment_terms,
        "schema_versions": versions,
    })


# ---------------------------------------------------------------------------
# Database Initialization
# ---------------------------------------------------------------------------

def _chmod_600(path: str) -> None:
    """Set file mode to 0o600 if the file exists. Silent on missing/permission errors."""
    try:
        if os.path.exists(path):
            os.chmod(path, 0o600)
    except OSError:
        pass


def chmod_db_files(db_path: str) -> None:
    """Set mode 600 on the SQLite DB file and its WAL/SHM siblings.

    Called from initialize_database, restore_database, and the every-action
    bootstrap to keep credentials + payroll data unreadable to other users
    on shared machines.
    """
    _chmod_600(db_path)
    _chmod_600(db_path + "-wal")
    _chmod_600(db_path + "-shm")


def _link_shared_library() -> None:
    """Symlink ``$ERPCLAW_HOME/lib`` -> bundled lib at the skill location.

    Single deterministic install-time setup. Replaces the v4.0.x self-heal
    copy mechanism, which the OpenClaw scanner flagged as self-modifying
    code. The symlink resolves to the version-current bundled lib for the
    lifetime of the install; on `clawhub update`, the symlink target is
    automatically the new lib (no re-link needed).

    The link target resolves through the single ERPCLAW_HOME point of truth
    (ADR-0017): unset ⇒ ~/.openclaw/erpclaw/lib, byte-identical to before;
    Hermes sets ERPCLAW_HOME to its skill-dir install so the link lands under
    the Hermes prefix. Existing dependent skills sys.path.insert the deployed
    path; the symlink keeps them working unchanged.
    """
    bundled_lib = os.path.join(os.path.dirname(os.path.abspath(__file__)), "lib")
    if not os.path.isdir(os.path.join(bundled_lib, "erpclaw_lib")):
        return
    target = os.path.join(
        os.path.expanduser(os.environ.get("ERPCLAW_HOME", "~/.openclaw/erpclaw")), "lib")
    os.makedirs(os.path.dirname(target), exist_ok=True)
    if os.path.exists(target) or os.path.islink(target):
        if os.path.islink(target) and os.path.realpath(target) == os.path.realpath(bundled_lib):
            return
        if os.path.islink(target):
            os.unlink(target)
        else:
            shutil.rmtree(target)
    os.symlink(bundled_lib, target)


def initialize_database(conn, args):
    """Initialize (or re-initialize) the ERPClaw database schema.

    Creates all tables, indexes, and constraints. Safe to run on an existing
    database — uses CREATE TABLE IF NOT EXISTS throughout. If --force is
    passed, drops and recreates the database from scratch.
    """
    _link_shared_library()
    db_path = args.db_path or DEFAULT_DB_PATH

    # Import the schema module (co-located in the same scripts/ directory)
    scripts_dir = os.path.dirname(os.path.abspath(__file__))
    sys.path.insert(0, scripts_dir)
    from init_schema import init_db, ALL_DDL_BLOCKS, redact_db_url

    force = getattr(args, "force", False) or getattr(args, "force_reinit", False)

    if force and os.path.exists(db_path):
        # Close existing connection before deleting
        if conn:
            conn.close()
        os.remove(db_path)

    # Run full schema initialization
    init_db(db_path)

    # Lock down DB file perms (covers data.sqlite, -wal, -shm)
    chmod_db_files(db_path)

    # Verify by reconnecting and counting
    verify_conn = sqlite3.connect(db_path)
    try:
        table_count = verify_conn.execute(
            "SELECT COUNT(*) FROM sqlite_master WHERE type='table'"
        ).fetchone()[0]
        index_count = verify_conn.execute(
            "SELECT COUNT(*) FROM sqlite_master WHERE type='index'"
        ).fetchone()[0]
        skill_count = verify_conn.execute(
            "SELECT COUNT(*) FROM schema_version"
        ).fetchone()[0]
    finally:
        verify_conn.close()

    # M39 (Wave G F6): catalog the foundation in its own erpclaw_module table.
    # ClawHub's post hook runs this action and nothing else, so this INSERT is
    # the ONLY chance a fresh install gets to be catalogued — without it
    # list-modules is empty, update-modules says "No modules to update", and the
    # dependency resolver drives the addon git-clone installer at the
    # foundation. Bookkeeping, never fatal: a failure here is REPORTED in the
    # response (never swallowed) rather than aborting a schema init that
    # otherwise succeeded, and `update-foundation` heals it on the next
    # reconcile through the same shared helper.
    module_row = {"ensured": False, "inserted": False, "reason": "not attempted"}
    try:
        from erpclaw_lib.foundation_registry import (
            ensure_foundation_module_row, load_foundation_entry,
        )
        registry_path = os.path.join(os.path.dirname(SKILL_DIR), "module_registry.json")
        row_conn = get_connection(db_path)
        try:
            module_row = ensure_foundation_module_row(
                row_conn,
                load_foundation_entry(registry_path),
                install_path=os.path.dirname(os.path.dirname(SKILL_DIR)),
            )
        finally:
            row_conn.close()
    except Exception as e:  # noqa: BLE001 — surfaced below, never fatal
        module_row = {"ensured": False, "inserted": False, "reason": str(e)}

    ok({
        "message": "Database initialized successfully",
        # F11: on PostgreSQL this value IS the connection URL, so the response
        # echoed the database password to stdout. Same redactor as the stderr
        # prints; a SQLite filesystem path passes through unchanged.
        "db_path": redact_db_url(db_path),
        "tables": table_count,
        "indexes": index_count,
        "skills_registered": skill_count,
        "journal_mode": "WAL",
        "foreign_keys": "ON",
        "reinitialized": force,
        "foundation_module_row": module_row,
    })


# ---------------------------------------------------------------------------
# Tutorial — demo company with guided next-steps
# ---------------------------------------------------------------------------

def tutorial(conn, args):
    """Create a demo company 'Acme Corp' with essential accounts and a fiscal year.

    Idempotent: if Acme Corp already exists, returns existing data with next-steps.
    This gives new users a working starting point for all skills.
    """
    now_year = datetime.now(timezone.utc).strftime("%Y")

    # Check if Acme Corp already exists
    tco = Table("company")
    q_existing = Q.from_(tco).select(tco.id).where(tco.name == "Acme Corp")
    existing = conn.execute(q_existing.get_sql()).fetchone()
    if existing:
        company_id = existing["id"]
        ta = Table("account")
        q_accts = Q.from_(ta).select(ta.id, ta.name, ta.account_type).where(ta.company_id == P())
        accounts = conn.execute(q_accts.get_sql(), (company_id,)).fetchall()
        ok({
            "message": "Acme Corp already exists. Demo data is ready.",
            "company_id": company_id,
            "accounts": len(accounts),
            "next_steps": _tutorial_next_steps(),
        })

    company_id = str(uuid.uuid4())

    # 1. Create Acme Corp
    tco2 = Table("company")
    q_ins_co = Q.into(tco2).columns(
        "id", "name", "abbr", "default_currency", "country", "fiscal_year_start_month"
    ).insert(P(), "Acme Corp", "AC", "USD", "United States", 1)
    conn.execute(q_ins_co.get_sql(), (company_id,))

    # 2. Create chart of accounts (minimal but functional)
    account_map = {}
    acct_defs = [
        # (name, account_number, account_type, root_type, is_group)
        ("Assets", "1000", None, "asset", 1),
        ("Bank Account", "1100", "bank", "asset", 0),
        ("Cash Account", "1200", "cash", "asset", 0),
        ("Accounts Receivable", "1300", "receivable", "asset", 0),
        ("Inventory", "1400", "stock", "asset", 0),
        ("Liabilities", "2000", None, "liability", 1),
        ("Accounts Payable", "2100", "payable", "liability", 0),
        ("Stock Received Not Billed", "2150", "stock_received_not_billed", "liability", 0),
        ("Sales Tax Payable", "2200", "tax", "liability", 0),
        ("Payroll Payable", "2300", "payroll_payable", "liability", 0),
        ("Federal Tax Payable", "2310", "tax", "liability", 0),
        ("Social Security Payable", "2320", "tax", "liability", 0),
        ("Income", "4000", None, "income", 1),
        ("Sales Revenue", "4100", "revenue", "income", 0),
        ("Expenses", "5000", None, "expense", 1),
        ("Cost of Goods Sold", "5100", "cost_of_goods_sold", "expense", 0),
        ("Operating Expenses", "5200", "expense", "expense", 0),
        ("Stock Adjustment", "5210", "stock_adjustment", "expense", 0),
        ("Salary Expense", "5300", "expense", "expense", 0),
        ("Equity", "3000", None, "equity", 1),
        ("Retained Earnings", "3100", "equity", "equity", 0),
    ]
    tacct = Table("account")
    q_ins_acct = Q.into(tacct).columns(
        "id", "name", "account_number", "account_type", "root_type",
        "is_group", "company_id", "balance_direction"
    ).insert(P(), P(), P(), P(), P(), P(), P(), P())
    for name, acct_num, acct_type, root_type, is_group in acct_defs:
        acct_id = str(uuid.uuid4())
        conn.execute(
            q_ins_acct.get_sql(),
            (acct_id, name, acct_num, acct_type, root_type,
             is_group, company_id,
             "debit_normal" if root_type in ("asset", "expense") else "credit_normal"),
        )
        account_map[acct_num] = acct_id

    # 3. Set company default accounts
    tco3 = Table("company")
    q_upd_co = (Q.update(tco3)
                .set(Field("default_receivable_account_id"), P())
                .set(Field("default_payable_account_id"), P())
                .set(Field("default_income_account_id"), P())
                .set(Field("default_expense_account_id"), P())
                .set(Field("default_bank_account_id"), P())
                .set(Field("default_cash_account_id"), P())
                .where(tco3.id == P()))
    conn.execute(
        q_upd_co.get_sql(),
        (account_map["1300"], account_map["2100"], account_map["4100"],
         account_map["5200"], account_map["1100"], account_map["1200"],
         company_id),
    )

    # 4. Create cost center
    cc_id = str(uuid.uuid4())
    tcc = Table("cost_center")
    q_ins_cc = Q.into(tcc).columns("id", "name", "company_id", "is_group").insert(P(), "Main", P(), 0)
    conn.execute(q_ins_cc.get_sql(), (cc_id, company_id))
    tco4 = Table("company")
    q_upd_cc = Q.update(tco4).set(Field("default_cost_center_id"), P()).where(tco4.id == P())
    conn.execute(q_upd_cc.get_sql(), (cc_id, company_id))

    # 5. Create fiscal year
    fy_id = str(uuid.uuid4())
    tfy = Table("fiscal_year")
    q_ins_fy = Q.into(tfy).columns(
        "id", "name", "start_date", "end_date", "company_id", "is_closed"
    ).insert(P(), P(), P(), P(), P(), 0)
    conn.execute(
        q_ins_fy.get_sql(),
        (fy_id, f"FY {now_year}", f"{now_year}-01-01", f"{now_year}-12-31", company_id),
    )

    audit(conn, "erpclaw-setup", "create", "company", company_id,
           new_values={"name": "Acme Corp", "accounts": len(acct_defs),
                       "fiscal_year": f"FY {now_year}"},
           description="Tutorial: created Acme Corp with demo data")
    conn.commit()

    ok({
        "message": "Acme Corp created with 15 accounts, 1 cost center, and a fiscal year.",
        "company_id": company_id,
        "company_name": "Acme Corp",
        "accounts_created": len(acct_defs),
        "fiscal_year": f"FY {now_year}",
        "cost_center": "Main",
        "next_steps": _tutorial_next_steps(),
    })


def _tutorial_next_steps():
    """Return guided next-steps for the tutorial."""
    return [
        {"step": 1, "skill": "erpclaw-gl",
         "action": "Your chart of accounts is ready. Try 'show my account balances' or 'list accounts'."},
        {"step": 2, "skill": "erpclaw-inventory",
         "action": "Add items and a warehouse: 'add item Widget, price $25' and 'add warehouse Main Warehouse'."},
        {"step": 3, "skill": "erpclaw-selling",
         "action": "Add a customer and create an invoice: 'add customer Stark Industries' then 'create invoice'."},
        {"step": 4, "skill": "erpclaw-payments",
         "action": "Record a payment: 'record payment from Stark Industries for $500'."},
        {"step": 5, "skill": "erpclaw-analytics",
         "action": "See the big picture: 'show me the executive dashboard' or 'how is business?'."},
    ]


# ---------------------------------------------------------------------------
# Cron / Maintenance Actions
# ---------------------------------------------------------------------------

def cleanup_backups(conn, args):
    """Clean up old database backups using a retention policy.

    Retention rules:
      - Keep the 7 most recent daily backups
      - Keep 4 weekly backups (oldest per week beyond the daily window)
      - Keep 12 monthly backups (oldest per month beyond the weekly window)
      - Delete everything else
    """
    backup_dir = BACKUP_DIR
    pattern = os.path.join(backup_dir, "erpclaw_backup_*.sqlite")
    files = sorted(glob_mod.glob(pattern), reverse=True)  # newest first

    if not files:
        ok({"kept": 0, "deleted": 0, "freed_bytes": 0})

    # Parse timestamps from filenames and build records
    backups = []
    for f in files:
        basename = os.path.basename(f)
        # erpclaw_backup_YYYYMMDD_HHMMSS.sqlite
        try:
            parts = basename.replace("erpclaw_backup_", "").replace(".sqlite", "")
            dt = datetime.strptime(parts, "%Y%m%d_%H%M%S")
            backups.append({"path": f, "dt": dt, "date": dt.date(),
                            "size": os.path.getsize(f)})
        except (ValueError, OSError):
            continue  # skip files that don't match expected format

    if not backups:
        ok({"kept": 0, "deleted": 0, "freed_bytes": 0})

    keep_set = set()

    # --- Daily: keep the 7 most recent unique dates ---
    seen_dates = set()
    daily_kept = 0
    for b in backups:
        if b["date"] not in seen_dates:
            seen_dates.add(b["date"])
            daily_kept += 1
        if daily_kept <= 7:
            keep_set.add(b["path"])

    # --- Weekly: 4 oldest-per-week beyond the daily window ---
    # Group remaining (not yet kept) by ISO week
    weekly_candidates = [b for b in backups if b["path"] not in keep_set]
    weeks_seen = {}  # (iso_year, iso_week) -> oldest backup
    for b in weekly_candidates:
        iso_year, iso_week, _ = b["date"].isocalendar()
        key = (iso_year, iso_week)
        if key not in weeks_seen or b["dt"] < weeks_seen[key]["dt"]:
            weeks_seen[key] = b

    # Sort weeks newest first and keep 4
    sorted_weeks = sorted(weeks_seen.keys(), reverse=True)
    for wk in sorted_weeks[:4]:
        keep_set.add(weeks_seen[wk]["path"])

    # --- Monthly: 12 oldest-per-month beyond daily+weekly ---
    monthly_candidates = [b for b in backups if b["path"] not in keep_set]
    months_seen = {}  # (year, month) -> oldest backup
    for b in monthly_candidates:
        key = (b["date"].year, b["date"].month)
        if key not in months_seen or b["dt"] < months_seen[key]["dt"]:
            months_seen[key] = b

    sorted_months = sorted(months_seen.keys(), reverse=True)
    for mo in sorted_months[:12]:
        keep_set.add(months_seen[mo]["path"])

    # --- Delete everything not in keep_set ---
    deleted = 0
    freed = 0
    for b in backups:
        if b["path"] not in keep_set:
            try:
                freed += b["size"]
                os.remove(b["path"])
                deleted += 1
            except OSError:
                pass  # file already gone or permission issue

    kept = len(backups) - deleted
    audit(conn, "erpclaw-setup", "cleanup", "backup", "system",
           new_values={"kept": kept, "deleted": deleted, "freed_bytes": freed},
           description=f"Backup cleanup: kept {kept}, deleted {deleted}")
    conn.commit()
    ok({"kept": kept, "deleted": deleted, "freed_bytes": freed})


def fetch_exchange_rates(conn, args):
    """Fetch latest exchange rates from frankfurter.dev (base=USD).

    For each rate returned, inserts or updates a row in exchange_rate with
    from_currency='USD', effective_date=today, source='api'.
    """
    url = "https://api.frankfurter.dev/latest?from=USD"
    today = date.today().isoformat()

    try:
        req = urllib.request.Request(url, headers={"User-Agent": "erpclaw/1.0"})
        with urllib.request.urlopen(req, timeout=15) as resp:
            data = json.loads(resp.read().decode("utf-8"))
    except (urllib.error.URLError, urllib.error.HTTPError, OSError) as e:
        err(f"Failed to fetch exchange rates: {e}",
             suggestion="Check internet connection or try again later.")

    rates = data.get("rates", {})
    if not rates:
        err("API returned no rates",
             suggestion="The API may be temporarily unavailable. Try again later.")

    count = 0
    for currency_code, rate_value in rates.items():
        rate_str = str(Decimal(str(rate_value)))

        # Check if a rate already exists for this pair + date
        ter = Table("exchange_rate")
        q_check = (Q.from_(ter).select(ter.id)
                   .where(ter.from_currency == "USD")
                   .where(ter.to_currency == P())
                   .where(ter.effective_date == P()))
        existing = conn.execute(q_check.get_sql(), (currency_code, today)).fetchone()

        if existing:
            q_upd = (Q.update(ter)
                     .set(ter.rate, P())
                     .set(ter.source, "api")
                     .set(Field("updated_at"), now())
                     .where(ter.id == P()))
            conn.execute(q_upd.get_sql(), (rate_str, existing["id"]))
        else:
            rate_id = str(uuid.uuid4())
            q_ins = Q.into(ter).columns(
                "id", "from_currency", "to_currency", "rate", "effective_date", "source"
            ).insert(P(), "USD", P(), P(), P(), "api")
            conn.execute(q_ins.get_sql(), (rate_id, currency_code, rate_str, today))
        count += 1

    audit(conn, "erpclaw-setup", "fetch", "exchange_rate", "system",
           new_values={"rates_updated": count, "source": "frankfurter.dev",
                       "date": today},
           description=f"Fetched {count} exchange rates from frankfurter.dev")
    conn.commit()
    ok({"rates_updated": count, "source": "frankfurter.dev",
         "base": "USD", "date": today})


# ---------------------------------------------------------------------------
# RBAC: Users, Roles, Permissions
# ---------------------------------------------------------------------------

def add_user(conn, args):
    """Create a new ERP user."""
    username = args.name  # reuse --name flag for username
    if not username:
        err("--name (username) is required", suggestion="Provide a unique username")
    email = getattr(args, "email", None)
    full_name = getattr(args, "full_name", None)
    company_id = args.company_id

    # Validate email format if provided
    _EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
    if email and not _EMAIL_RE.match(email):
        err(f"Invalid email format for --email: '{email}'")

    # Check uniqueness
    tu = Table("erp_user")
    q_check = Q.from_(tu).select(tu.id).where(tu.username == P())
    existing = conn.execute(q_check.get_sql(), (username,)).fetchone()
    if existing:
        err(f"Username '{username}' already exists",
             suggestion="Choose a different username")

    user_id = str(uuid.uuid4())
    company_ids = json.dumps([company_id]) if company_id else None

    q_ins = Q.into(tu).columns(
        "id", "username", "email", "full_name", "company_ids"
    ).insert(P(), P(), P(), P(), P())
    conn.execute(
        q_ins.get_sql(),
        (user_id, username, email, full_name, company_ids),
    )
    audit(conn, "erpclaw-setup", "add-user", "erp_user", user_id,
           new_values={"username": username, "email": email})
    conn.commit()
    ok({"user_id": user_id, "username": username})


def update_user(conn, args):
    """Update an existing ERP user."""
    user_id = getattr(args, "user_id", None)
    if not user_id:
        err("--user-id is required")

    tu = Table("erp_user")
    q_get = Q.from_(tu).select(tu.star).where(tu.id == P())
    user = conn.execute(q_get.get_sql(), (user_id,)).fetchone()
    if not user:
        err("User not found")

    updates = {}
    if args.name:
        updates["username"] = args.name
    if getattr(args, "email", None):
        updates["email"] = args.email
    if getattr(args, "full_name", None):
        updates["full_name"] = args.full_name
    if getattr(args, "user_status", None):
        if args.user_status not in ("active", "disabled", "locked"):
            err("--user-status must be active, disabled, or locked")
        updates["status"] = args.user_status
    if args.company_id:
        # Append company to existing list
        existing = json.loads(user["company_ids"] or "[]")
        if args.company_id not in existing:
            existing.append(args.company_id)
        updates["company_ids"] = json.dumps(existing)

    if not updates:
        err("No fields to update")

    tu2 = Table("erp_user")
    qu = Q.update(tu2)
    for k in updates:
        qu = qu.set(Field(k), P())
    qu = qu.set(Field("updated_at"), now())
    qu = qu.where(tu2.id == P())
    vals = list(updates.values()) + [user_id]
    conn.execute(qu.get_sql(), vals)
    audit(conn, "erpclaw-setup", "update-user", "erp_user", user_id,
           old_values=dict(user), new_values=updates)
    conn.commit()
    ok({"user_id": user_id, "updated_fields": list(updates.keys())})


def list_users(conn, args):
    """List all ERP users."""
    limit = args.limit or 50
    offset = args.offset or 0
    tu = Table("erp_user")
    q = (Q.from_(tu)
         .select(tu.id, tu.username, tu.email, tu.full_name, tu.status, tu.company_ids, tu.created_at)
         .orderby(tu.username)
         .limit(limit + 1)
         .offset(offset))
    rows = conn.execute(q.get_sql()).fetchall()
    users = [dict(r) for r in rows[:limit]]
    ok({"users": users, "count": len(users),
         "has_more": len(rows) > limit})


def get_user(conn, args):
    """Get details of a specific ERP user including roles."""
    user_id = getattr(args, "user_id", None)
    if not user_id:
        err("--user-id is required")

    tu = Table("erp_user")
    q_user = Q.from_(tu).select(tu.star).where(tu.id == P())
    user = conn.execute(q_user.get_sql(), (user_id,)).fetchone()
    if not user:
        err("User not found")

    tur = Table("user_role").as_("ur")
    tr = Table("role").as_("r")
    tco = Table("company").as_("c")
    q_roles = (Q.from_(tur)
               .join(tr).on(tr.id == tur.role_id)
               .left_join(tco).on(tco.id == tur.company_id)
               .select(
                   tr.name.as_("role_name"),
                   tur.company_id,
                   tco.name.as_("company_name"),
               )
               .where(tur.user_id == P())
               .orderby(tr.name))
    roles = conn.execute(q_roles.get_sql(), (user_id,)).fetchall()

    result = dict(user)
    result["roles"] = [dict(r) for r in roles]
    ok(result)


def add_role(conn, args):
    """Create a custom (non-system) role."""
    role_name = args.name
    if not role_name:
        err("--name is required")

    tr = Table("role")
    q_check = Q.from_(tr).select(tr.id).where(tr.name == P())
    existing = conn.execute(q_check.get_sql(), (role_name,)).fetchone()
    if existing:
        err(f"Role '{role_name}' already exists")

    role_id = str(uuid.uuid4())
    q_ins = Q.into(tr).columns("id", "name", "description", "is_system").insert(P(), P(), P(), 0)
    conn.execute(q_ins.get_sql(), (role_id, role_name, args.description))
    audit(conn, "erpclaw-setup", "add-role", "role", role_id,
           new_values={"name": role_name})
    conn.commit()
    ok({"role_id": role_id, "name": role_name})


def list_roles(conn, args):
    """List all roles."""
    tr = Table("role").as_("r")
    tur = Table("user_role").as_("ur")
    q = (Q.from_(tr)
         .left_join(tur).on(tur.role_id == tr.id)
         .select(tr.id, tr.name, tr.description, tr.is_system,
                 fn.Count(tur.id).as_("user_count"))
         .groupby(tr.id)
         .orderby(tr.is_system, order=Order.desc)
         .orderby(tr.name))
    rows = conn.execute(q.get_sql()).fetchall()
    ok({"roles": [dict(r) for r in rows], "count": len(rows)})


def assign_role(conn, args):
    """Assign a role to a user (optionally company-scoped)."""
    user_id = getattr(args, "user_id", None)
    role_name = getattr(args, "role_name", None)
    if not user_id:
        err("--user-id is required")
    if not role_name:
        err("--role-name is required")

    tu = Table("erp_user")
    q_user = Q.from_(tu).select(tu.id).where(tu.id == P())
    user = conn.execute(q_user.get_sql(), (user_id,)).fetchone()
    if not user:
        err("User not found")

    tr = Table("role")
    q_role = Q.from_(tr).select(tr.id).where(tr.name == P())
    role = conn.execute(q_role.get_sql(), (role_name,)).fetchone()
    if not role:
        err(f"Role '{role_name}' not found",
             suggestion="Use list-roles to see available roles")

    role_id = role["id"]
    company_id = args.company_id  # None = global assignment

    # Check if already assigned — use raw SQL for IS ? (NULL-safe comparison)
    existing = conn.execute(
        "SELECT id FROM user_role WHERE user_id = ? AND role_id = ? AND company_id IS ?",
        (user_id, role_id, company_id),
    ).fetchone()
    if existing:
        err(f"Role '{role_name}' already assigned to this user")

    ur_id = str(uuid.uuid4())
    tur = Table("user_role")
    q_ins = Q.into(tur).columns("id", "user_id", "role_id", "company_id").insert(P(), P(), P(), P())
    conn.execute(q_ins.get_sql(), (ur_id, user_id, role_id, company_id))
    audit(conn, "erpclaw-setup", "assign-role", "user_role", ur_id,
           new_values={"user_id": user_id, "role_name": role_name,
                       "company_id": company_id})
    conn.commit()
    ok({"user_role_id": ur_id, "role_name": role_name,
         "company_id": company_id})


def revoke_role(conn, args):
    """Remove a role from a user."""
    user_id = getattr(args, "user_id", None)
    role_name = getattr(args, "role_name", None)
    if not user_id:
        err("--user-id is required")
    if not role_name:
        err("--role-name is required")

    tr = Table("role")
    q_role = Q.from_(tr).select(tr.id).where(tr.name == P())
    role = conn.execute(q_role.get_sql(), (role_name,)).fetchone()
    if not role:
        err(f"Role '{role_name}' not found")

    company_id = args.company_id
    # Use raw SQL for IS ? (NULL-safe comparison not supported by PyPika for SQLite)
    deleted = conn.execute(
        "DELETE FROM user_role WHERE user_id = ? AND role_id = ? AND company_id IS ?",
        (user_id, role["id"], company_id),
    )
    if deleted.rowcount == 0:
        err(f"Role '{role_name}' not assigned to this user")

    audit(conn, "erpclaw-setup", "revoke-role", "user_role", user_id,
           old_values={"role_name": role_name, "company_id": company_id})
    conn.commit()
    ok({"revoked": role_name, "user_id": user_id})


def set_password(conn, args):
    """Set web login password for a user."""
    user_id = getattr(args, "user_id", None)
    password = getattr(args, "password", None)
    if not user_id:
        err("--user-id is required")
    if not password:
        err("--password is required")
    if len(password) < 8:
        err("Password must be at least 8 characters")

    tu = Table("erp_user")
    q_get = Q.from_(tu).select(tu.id, tu.username).where(tu.id == P())
    user = conn.execute(q_get.get_sql(), (user_id,)).fetchone()
    if not user:
        err("User not found")

    from erpclaw_lib.passwords import hash_password
    pw_hash = hash_password(password)
    q_upd = (Q.update(tu)
             .set(Field("password_hash"), P())
             .set(Field("updated_at"), now())
             .where(tu.id == P()))
    conn.execute(q_upd.get_sql(), (pw_hash, user_id))
    audit(conn, "erpclaw-setup", "set-password", "erp_user", user_id,
           description="Web password set")
    conn.commit()
    username = user["username"] if isinstance(user, dict) else user[1]
    ok({"user_id": user_id, "username": username, "message": "Password set successfully"})


def link_telegram_user(conn, args):
    """Link a Telegram user ID to an ERP user account."""
    user_id = getattr(args, "user_id", None)
    telegram_user_id = getattr(args, "telegram_user_id", None)
    if not user_id:
        err("--user-id is required")
    if not telegram_user_id:
        err("--telegram-user-id is required")

    tu = Table("erp_user")
    q_get = Q.from_(tu).select(tu.id, tu.username).where(tu.id == P())
    user = conn.execute(q_get.get_sql(), (user_id,)).fetchone()
    if not user:
        err("User not found")

    # Check if telegram_user_id already linked to another user
    q_check = Q.from_(tu).select(tu.id, tu.username).where(tu.telegram_user_id == P())
    existing = conn.execute(q_check.get_sql(), (str(telegram_user_id),)).fetchone()
    if existing:
        ex_id = existing["id"] if isinstance(existing, dict) else existing[0]
        if ex_id != user_id:
            err(f"Telegram user {telegram_user_id} is already linked to another account")

    q_upd = (Q.update(tu)
             .set(Field("telegram_user_id"), P())
             .set(Field("updated_at"), now())
             .where(tu.id == P()))
    conn.execute(q_upd.get_sql(), (str(telegram_user_id), user_id))
    audit(conn, "erpclaw-setup", "link-telegram-user", "erp_user", user_id,
          new_values={"telegram_user_id": str(telegram_user_id)})
    conn.commit()

    username = user["username"] if isinstance(user, dict) else user[1]
    ok({"user_id": user_id, "username": username,
        "telegram_user_id": str(telegram_user_id), "linked": True})


def unlink_telegram_user(conn, args):
    """Remove Telegram user ID link from an ERP user account."""
    telegram_user_id = getattr(args, "telegram_user_id", None)
    if not telegram_user_id:
        err("--telegram-user-id is required")

    tu = Table("erp_user")
    q_get = Q.from_(tu).select(tu.id, tu.username).where(tu.telegram_user_id == P())
    user = conn.execute(q_get.get_sql(), (str(telegram_user_id),)).fetchone()
    if not user:
        err(f"No user linked to Telegram user {telegram_user_id}")

    user_id = user["id"] if isinstance(user, dict) else user[0]
    q_upd = (Q.update(tu)
             .set(Field("telegram_user_id"), None)
             .set(Field("updated_at"), now())
             .where(tu.id == P()))
    conn.execute(q_upd.get_sql(), (user_id,))
    audit(conn, "erpclaw-setup", "unlink-telegram-user", "erp_user", user_id,
          old_values={"telegram_user_id": str(telegram_user_id)})
    conn.commit()

    ok({"user_id": user_id, "telegram_user_id": str(telegram_user_id), "unlinked": True})


def check_telegram_permission(conn, args):
    """Check if a Telegram user has permission for a specific skill action."""
    telegram_user_id = getattr(args, "telegram_user_id", None)
    skill = getattr(args, "skill", None)
    action_name = getattr(args, "check_action", None)
    if not telegram_user_id:
        err("--telegram-user-id is required")
    if not skill:
        err("--skill is required")
    if not action_name:
        err("--check-action is required")

    from erpclaw_lib.rbac import resolve_telegram_user_id, check_permission

    user_id = resolve_telegram_user_id(conn, telegram_user_id)
    if not user_id:
        ok({"allowed": False, "reason": "not_linked",
            "telegram_user_id": str(telegram_user_id)})
        return

    allowed = check_permission(conn, user_id, skill, action_name)
    ok({"allowed": allowed, "user_id": user_id,
        "skill": skill, "action": action_name,
        "telegram_user_id": str(telegram_user_id)})


def seed_permissions(conn, args):
    """Seed default role permissions from the shared RBAC library."""
    from erpclaw_lib.rbac import seed_role_permissions
    from erpclaw_lib.args import SafeArgumentParser, check_unknown_args
    seed_role_permissions(conn)
    t = Table("role_permission")
    q = Q.from_(t).select(fn.Count("*").as_("cnt"))
    count = conn.execute(q.get_sql()).fetchone()
    ok({"permissions_seeded": count["cnt"]})


# ---------------------------------------------------------------------------
# Onboarding Wizard
# ---------------------------------------------------------------------------

ONBOARDING_STATE_DIR = os.path.expanduser(
    os.environ.get("ERPCLAW_HOME", "~/.openclaw/erpclaw"))
ONBOARDING_STATE_FILE = os.path.join(ONBOARDING_STATE_DIR, "onboarding_state.json")

VALID_CURRENCIES = ["USD", "EUR", "GBP", "CAD", "INR", "SGD", "AED"]


def _load_onboarding_state():
    """Load onboarding state from file, return default if not found."""
    if os.path.exists(ONBOARDING_STATE_FILE):
        try:
            with open(ONBOARDING_STATE_FILE, "r") as f:
                return json.load(f)
        except (json.JSONDecodeError, IOError):
            pass
    return {"step": 1, "data": {}, "completed": False}


def _save_onboarding_state(state):
    """Save onboarding state to file."""
    os.makedirs(ONBOARDING_STATE_DIR, exist_ok=True)
    with open(ONBOARDING_STATE_FILE, "w") as f:
        json.dump(state, f)


def _clear_onboarding_state():
    """Remove onboarding state file."""
    if os.path.exists(ONBOARDING_STATE_FILE):
        os.remove(ONBOARDING_STATE_FILE)


def onboarding_step(conn, args):
    """Interactive onboarding wizard — state-machine driven.

    Each call advances one step. Pass --answer with the user's response.
    Pass --reset to restart the wizard from step 1.
    """
    if args.reset:
        _clear_onboarding_state()
        state = {"step": 1, "data": {}, "completed": False}
        _save_onboarding_state(state)
        ok({"step": 1, "completed": False,
            "prompt": "Welcome to ERPClaw! Let's set up your company.\n\nWhat's your company name?",
            "options": [], "field": "company_name"})
        return

    state = _load_onboarding_state()

    # If already completed, report that
    if state.get("completed"):
        ok({"step": 5, "completed": True,
            "prompt": "Onboarding already completed! Your company is set up and ready to use.",
            "company_name": state["data"].get("company_name", ""),
            "suggestion": "Try 'list companies' or 'show status' to see your setup."})
        return

    answer = (args.answer or "").strip()
    step = state["step"]

    # Step 1: Company name
    if step == 1:
        if not answer:
            ok({"step": 1, "completed": False,
                "prompt": "What's your company name?",
                "options": [], "field": "company_name"})
            return
        state["data"]["company_name"] = answer
        state["step"] = 2
        _save_onboarding_state(state)
        ok({"step": 2, "completed": False,
            "prompt": f"Great! Company name: {answer}\n\nWhat currency? (USD/EUR/GBP/CAD/INR/SGD/AED)",
            "options": VALID_CURRENCIES, "field": "currency"})
        return

    # Step 2: Currency
    if step == 2:
        if not answer:
            ok({"step": 2, "completed": False,
                "prompt": "What currency? (USD/EUR/GBP/CAD/INR/SGD/AED)",
                "options": VALID_CURRENCIES, "field": "currency"})
            return
        currency = answer.upper().strip()
        if currency not in VALID_CURRENCIES:
            ok({"step": 2, "completed": False,
                "prompt": f"'{answer}' is not a supported currency. Please choose one of: {', '.join(VALID_CURRENCIES)}",
                "options": VALID_CURRENCIES, "field": "currency",
                "error": "invalid_currency"})
            return
        state["data"]["currency"] = currency
        state["step"] = 3
        _save_onboarding_state(state)
        ok({"step": 3, "completed": False,
            "prompt": f"Currency set to {currency}.\n\nFiscal year start month? (1-12, default: 1 for January)",
            "options": ["1", "4", "7", "10"], "field": "fiscal_month"})
        return

    # Step 3: Fiscal year start month
    if step == 3:
        if not answer:
            ok({"step": 3, "completed": False,
                "prompt": "Fiscal year start month? (1-12, default: 1 for January)",
                "options": ["1", "4", "7", "10"], "field": "fiscal_month"})
            return
        try:
            month = int(answer)
            if month < 1 or month > 12:
                raise ValueError()
        except ValueError:
            ok({"step": 3, "completed": False,
                "prompt": f"'{answer}' is not a valid month. Please enter a number from 1 to 12.",
                "options": ["1", "4", "7", "10"], "field": "fiscal_month",
                "error": "invalid_month"})
            return
        state["data"]["fiscal_month"] = month
        state["step"] = 4
        _save_onboarding_state(state)
        ok({"step": 4, "completed": False,
            "prompt": "Load demo data? This creates sample customers, items, invoices, and more. (yes/no)",
            "options": ["yes", "no"], "field": "load_demo"})
        return

    # Step 4: Demo data
    if step == 4:
        if not answer:
            ok({"step": 4, "completed": False,
                "prompt": "Load demo data? (yes/no)",
                "options": ["yes", "no"], "field": "load_demo"})
            return
        load_demo = answer.lower().strip() in ("yes", "y", "true", "1")
        state["data"]["load_demo"] = load_demo
        state["step"] = 5

        # Execute: create company, seed defaults, setup chart of accounts
        data = state["data"]
        company_name = data["company_name"]
        currency = data.get("currency", "USD")
        fiscal_month = data.get("fiscal_month", 1)
        results = {"steps_completed": [], "steps_failed": []}

        def _record_step(step_name, proc):
            """Record a subprocess setup step outcome. A zero exit appends to
            steps_completed; a non-zero exit is surfaced in steps_failed and on
            stderr rather than being reported as completed. subprocess.run is
            called without check=True, so without this a failed step would
            silently look successful."""
            if proc.returncode == 0:
                results["steps_completed"].append(step_name)
            else:
                detail = (proc.stderr or proc.stdout or "").strip()[:300]
                results["steps_failed"].append(
                    {"step": step_name, "error": detail or f"exit {proc.returncode}"})
                print(f"WARN: onboarding step '{step_name}' exited "
                      f"{proc.returncode}: {detail[:200]}", file=sys.stderr)

        # Step A: Create company
        import subprocess
        script_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "db_query.py")

        try:
            result = subprocess.run(
                [sys.executable, script_path,
                 "--action", "setup-company",
                 "--name", company_name,
                 "--currency", currency,
                 "--fiscal-year-start-month", str(fiscal_month)],
                capture_output=True, text=True, timeout=30
            )
            company_result = json.loads(result.stdout) if result.stdout.strip() else {}
            if "error" in company_result:
                _save_onboarding_state(state)
                ok({"step": 5, "completed": False,
                    "error": f"Company creation failed: {company_result['error']}",
                    "prompt": "There was an error creating your company. Try again with 'onboarding-step --reset'."})
                return
            company_id = company_result.get("company_id", "")
            results["steps_completed"].append("setup-company")
            results["company_id"] = company_id
        except Exception as e:
            _save_onboarding_state(state)
            ok({"step": 5, "completed": False,
                "error": f"Company creation failed: {str(e)}",
                "prompt": "There was an error. Try again with 'onboarding-step --reset'."})
            return

        # Step B: Seed defaults
        try:
            result = subprocess.run(
                [sys.executable, script_path,
                 "--action", "seed-defaults",
                 "--company-id", company_id],
                capture_output=True, text=True, timeout=30
            )
            _record_step("seed-defaults", result)
        except Exception as e:
            results["steps_failed"].append({"step": "seed-defaults", "error": str(e)})
            print(f"WARN: onboarding step 'seed-defaults' failed: {e}", file=sys.stderr)

        # Step C: Setup chart of accounts (via erpclaw-gl if available)
        gl_script = os.path.join(
            os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
            "erpclaw-gl", "scripts", "db_query.py"
        )
        if os.path.exists(gl_script):
            try:
                result = subprocess.run(
                    [sys.executable, gl_script,
                     "--action", "setup-chart-of-accounts",
                     "--company-id", company_id,
                     "--standard", "us_gaap"],
                    capture_output=True, text=True, timeout=30
                )
                _record_step("setup-chart-of-accounts", result)
            except Exception as e:
                results["steps_failed"].append({"step": "setup-chart-of-accounts", "error": str(e)})
                print(f"WARN: onboarding step 'setup-chart-of-accounts' failed: {e}", file=sys.stderr)

        # Step D: Load demo data if requested (via erpclaw meta-package)
        if load_demo:
            erpclaw_script = os.path.join(
                os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))),
                "erpclaw", "scripts", "db_query.py"
            )
            if os.path.exists(erpclaw_script):
                try:
                    result = subprocess.run(
                        [sys.executable, erpclaw_script,
                         "--action", "seed-demo-data"],
                        capture_output=True, text=True, timeout=120
                    )
                    _record_step("seed-demo-data", result)
                except Exception as e:
                    results["steps_failed"].append({"step": "seed-demo-data", "error": str(e)})
                    print(f"WARN: onboarding step 'seed-demo-data' failed: {e}", file=sys.stderr)

        state["completed"] = True
        _save_onboarding_state(state)

        month_names = ["January", "February", "March", "April", "May", "June",
                       "July", "August", "September", "October", "November", "December"]
        fiscal_name = month_names[fiscal_month - 1]

        failed_note = ""
        if results["steps_failed"]:
            failed_names = ", ".join(s["step"] for s in results["steps_failed"])
            failed_note = (f"Some steps did not complete: {failed_names}. "
                           f"Your company was created; you can re-run those steps "
                           f"or check the logs.\n\n")

        summary = (
            f"Setup complete! Here's your configuration:\n\n"
            f"  Company: {company_name}\n"
            f"  Currency: {currency}\n"
            f"  Fiscal Year Start: {fiscal_name}\n"
            f"  Demo Data: {'Loaded' if load_demo else 'Skipped'}\n\n"
            f"Steps completed: {', '.join(results['steps_completed'])}\n\n"
            f"{failed_note}"
            f"You're ready to go! Try:\n"
            f"  - 'list customers' to see your customer data\n"
            f"  - 'show trial balance' to view your financials\n"
            f"  - 'create an invoice' to start selling"
        )

        ok({"step": 5, "completed": True,
            "prompt": summary,
            "company_name": company_name,
            "company_id": company_id,
            "currency": currency,
            "fiscal_month": fiscal_month,
            "load_demo": load_demo,
            "results": results})
        return

    # Shouldn't reach here
    ok({"step": step, "completed": False,
        "prompt": "Unknown state. Use 'onboarding-step --reset' to restart.",
        "error": "invalid_state"})


# ---------------------------------------------------------------------------
# Action Router
# ---------------------------------------------------------------------------

def import_master_key_from_backup_action(conn, args):
    """Extract the wrapped column-encryption master key from a backup file
    and install it at ~/.config/erpclaw/master.key.

    Use case: cross-machine restore. After a backup taken on Machine A is
    transferred to Machine B, Machine B has no column-encryption master
    key and cannot decrypt sensitive fields (employee.ssn, etc.). This
    action reads the wrapped master key from the backup's ECRYPT02 header,
    unwraps it with the backup passphrase, and writes it to the standard
    master-key location on Machine B.

    Required: --backup-path. Passphrase via --passphrase, --passphrase-from-stdin,
    or --passphrase-from-env <VAR>.

    Refuses to overwrite an existing master key (intentional safety: the
    user must explicitly delete the existing key first to avoid clobbering
    encrypted data on the current machine).
    """
    backup_path = getattr(args, "backup_path", None)
    if not backup_path:
        err("--backup-path is required")
    if not os.path.isfile(backup_path):
        err(f"backup file not found: {backup_path}")

    # Resolve passphrase
    passphrase = None
    if getattr(args, "passphrase_from_stdin", False):
        passphrase = sys.stdin.read().strip()
    elif getattr(args, "passphrase_from_env", None):
        passphrase = os.environ.get(args.passphrase_from_env)
        if not passphrase:
            err(f"env var {args.passphrase_from_env} is empty or unset")
    elif getattr(args, "passphrase", None):
        passphrase = args.passphrase
    else:
        err(
            "passphrase required: --passphrase <pw> (avoid in shared shell history), "
            "--passphrase-from-stdin (recommended for automation), or "
            "--passphrase-from-env <VAR>"
        )
    if not passphrase:
        err("passphrase is empty")

    # Read just the ECRYPT02 header to extract wrapped master key
    from erpclaw_lib.crypto import _unpack_header_v2, unwrap_master_key, ECRYPT02_MAGIC
    try:
        with open(backup_path, "rb") as fh:
            magic_peek = fh.read(len(ECRYPT02_MAGIC))
            if magic_peek != ECRYPT02_MAGIC:
                err(
                    f"backup is not an ECRYPT02-format file (does not carry a "
                    f"wrapped master key). Got magic {magic_peek!r}. Legacy "
                    f"ECRYPT01 backups do not embed master keys; export the "
                    f"master key from the source machine manually."
                )
            fh.seek(0)
            iterations, salt, nonce_prefix, wrapped = _unpack_header_v2(fh)
    except Exception as exc:
        err(f"failed to read backup header: {exc}")

    if not wrapped:
        err(
            "backup does not carry a wrapped master key. The source machine "
            "either had no master key when the backup was taken, or the backup "
            "was created with a foundation version older than v4.1.3."
        )

    # Unwrap with passphrase
    try:
        master_key = unwrap_master_key(wrapped, passphrase)
    except Exception:
        err("passphrase did not match the wrapping passphrase used at backup time")

    # Install
    from erpclaw_lib.master_key import master_key_exists, import_master_key, MASTER_KEY_PATH
    force = getattr(args, "force", False)
    if master_key_exists() and not force:
        err(
            f"a master key already exists at {MASTER_KEY_PATH}. Refusing to "
            f"overwrite without --force. If you are sure the new key replaces "
            f"the existing one (e.g., the existing key is from an unrelated "
            f"install), pass --force. WARNING: overwriting will make existing "
            f"encrypted data unreadable if it was encrypted with the old key."
        )

    try:
        if force and master_key_exists():
            os.remove(MASTER_KEY_PATH)
        import_master_key(master_key)
    except Exception as exc:
        err(f"failed to install master key: {exc}")

    ok({
        "message": "Master key extracted from backup and installed.",
        "master_key_path": MASTER_KEY_PATH,
        "next": "run restore-database to restore the data; encrypted columns will now be readable",
    })


def set_credential_action(conn, args):
    """Store an integration credential in the encrypted credentials file.

    Reads the credential value from --value (CLI), --from-stdin (pipe), or
    --from-env (env var name). Never accepts the value via --api-key.
    """
    from erpclaw_lib import credentials as creds
    integration = getattr(args, "integration", None)
    if not integration:
        err("--integration is required")
    value = None
    if getattr(args, "from_stdin", False):
        value = sys.stdin.read().strip()
    elif getattr(args, "from_env", None):
        value = os.environ.get(args.from_env)
        if not value:
            err(f"env var {args.from_env} is empty or unset")
    elif getattr(args, "value", None):
        value = args.value
    else:
        err(
            "credential value required: pass --value <v> (avoid for shared "
            "history), or pipe via --from-stdin, or use --from-env <VAR>"
        )
    if not value:
        err("credential value is empty")
    creds.set_credential(integration, value)
    ok({
        "message": f"Credential '{integration}' stored.",
        "integration": integration,
        "credentials_file": creds.CREDENTIALS_PATH,
    })


def get_credential_action(conn, args):
    """Return whether a credential exists. Never returns the value itself."""
    from erpclaw_lib import credentials as creds
    integration = getattr(args, "integration", None)
    if not integration:
        err("--integration is required")
    val = creds.get_credential(integration)
    ok({
        "integration": integration,
        "exists": val is not None,
        "redacted_preview": (val[:4] + "..." + val[-4:]) if val and len(val) >= 12 else None,
    })


def list_credentials_action(conn, args):
    """List integration names that have credentials stored. Never lists values."""
    from erpclaw_lib import credentials as creds
    ok({"integrations": creds.list_credentials()})


def delete_credential_action(conn, args):
    """Remove a stored credential."""
    from erpclaw_lib import credentials as creds
    integration = getattr(args, "integration", None)
    if not integration:
        err("--integration is required")
    deleted = creds.delete_credential(integration)
    ok({
        "integration": integration,
        "deleted": deleted,
        "message": "deleted" if deleted else "no credential found for that integration",
    })


def migrate_credentials_action(conn, args):
    """Migrate plaintext credentials from per-addon DB tables into the encrypted file.

    Detects known credential-storing tables (Stripe, Shopify) and offers to
    import their plaintext keys into the encrypted credentials store. Use
    --dry-run to preview without writing.
    """
    from erpclaw_lib import credentials as creds
    dry_run = getattr(args, "dry_run", False)
    moved = []
    skipped = []
    cursor = conn.cursor()

    # Stripe addon: stripe_account.api_key
    try:
        rows = cursor.execute(
            "SELECT account_id, api_key FROM stripe_account WHERE api_key IS NOT NULL"
        ).fetchall()
        for r in rows:
            integration = f"stripe:{r[0]}" if len(rows) > 1 else "stripe"
            if creds.get_credential(integration):
                skipped.append({"integration": integration, "reason": "already in encrypted store"})
                continue
            if not dry_run:
                creds.set_credential(integration, r[1])
                cursor.execute(
                    "UPDATE stripe_account SET api_key = NULL WHERE account_id = ?",
                    (r[0],)
                )
            moved.append({"integration": integration, "source": "stripe_account.api_key"})
    except Exception:
        pass  # table doesn't exist; no Stripe addon installed

    if not dry_run:
        conn.commit()

    ok({
        "dry_run": dry_run,
        "moved": moved,
        "skipped": skipped,
        "next": "remove `--api-key` from any deploy/CI scripts; use `set-credential` instead",
    })


# ---------------------------------------------------------------------------
# Type/status registry administration (M0 additive slice)
# The registries (account_type_registry, voucher_type_registry,
# party_type_registry, asset_status_registry) are the runtime source of truth
# for the type/status values that used to be hardcoded CHECK constraints.
# These actions let an admin extend + inspect + soft-disable them at runtime.
# ---------------------------------------------------------------------------

def add_account_type(conn, args):
    """Register a new account_type so accounts can use it (M0)."""
    at = (args.account_type or "").strip()
    if not at:
        err("--account-type is required")
    label = args.label or at.replace("_", " ").title()
    skill = args.skill_name or "custom"
    try:
        conn.execute(
            "INSERT INTO account_type_registry (account_type, skill_name, label, is_active) "
            "VALUES (?, ?, ?, 1)", (at, skill, label))
        audit(conn, "erpclaw-setup", "create", "account_type_registry", at,
              new_values={"account_type": at, "label": label})
        conn.commit()
    except sqlite3.IntegrityError:
        err(f"account_type '{at}' is already registered")
    ok({"result": "registered", "account_type": at, "label": label})


def list_account_types(conn, args):
    """List registered account types (active only unless --include-inactive)."""
    t = Table("account_type_registry")
    q = Q.from_(t).select(t.star)
    if not args.include_inactive:
        q = q.where(t.is_active == 1)
    rows = conn.execute(q.orderby(t.account_type).get_sql()).fetchall()
    ok({"account_types": [row_to_dict(r) for r in rows], "count": len(rows)})


def deactivate_account_type(conn, args):
    """Soft-disable an account_type (is_active=0). Blocked if any account uses it."""
    at = (args.account_type or "").strip()
    if not at:
        err("--account-type is required")
    in_use = conn.execute(
        "SELECT COUNT(*) AS c FROM account WHERE account_type = ?", (at,)).fetchone()["c"]
    if in_use:
        err(f"Cannot deactivate account_type '{at}': {in_use} account(s) still use it.")
    conn.execute("UPDATE account_type_registry SET is_active = 0 WHERE account_type = ?", (at,))
    if conn.total_changes == 0:
        err(f"account_type '{at}' is not registered")
    audit(conn, "erpclaw-setup", "update", "account_type_registry", at,
          new_values={"is_active": 0})
    conn.commit()
    ok({"result": "deactivated", "account_type": at})


_VOUCHER_TARGETS = ("gl_entry", "stock_ledger_entry", "payment_allocation")


def add_voucher_type(conn, args):
    """Register a new voucher_type for a target table (M0)."""
    vt = (args.voucher_type or "").strip()
    target = (args.target_table or "").strip()
    if not vt:
        err("--voucher-type is required")
    if target not in _VOUCHER_TARGETS:
        err(f"--target-table must be one of: {', '.join(_VOUCHER_TARGETS)}")
    label = args.label or vt.replace("_", " ").title()
    skill = args.skill_name or "custom"
    try:
        conn.execute(
            "INSERT INTO voucher_type_registry (voucher_type, skill_name, label, target_table, is_active) "
            "VALUES (?, ?, ?, ?, 1)", (vt, skill, label, target))
        audit(conn, "erpclaw-setup", "create", "voucher_type_registry", f"{vt}/{target}",
              new_values={"voucher_type": vt, "target_table": target, "label": label})
        conn.commit()
    except sqlite3.IntegrityError:
        err(f"voucher_type '{vt}' is already registered for {target}")
    ok({"result": "registered", "voucher_type": vt, "target_table": target, "label": label})


def list_voucher_types(conn, args):
    """List registered voucher types (optionally filtered by --target-table)."""
    t = Table("voucher_type_registry")
    q = Q.from_(t).select(t.star)
    if not args.include_inactive:
        q = q.where(t.is_active == 1)
    if args.target_table:
        q = q.where(t.target_table == P())
        params = (args.target_table,)
    else:
        params = ()
    rows = conn.execute(q.orderby(t.target_table).orderby(t.voucher_type).get_sql(), params).fetchall()
    ok({"voucher_types": [row_to_dict(r) for r in rows], "count": len(rows)})


def deactivate_voucher_type(conn, args):
    """Soft-disable a voucher_type for a target table. Blocked if live rows use it."""
    vt = (args.voucher_type or "").strip()
    target = (args.target_table or "").strip()
    if not vt:
        err("--voucher-type is required")
    if target not in _VOUCHER_TARGETS:
        err(f"--target-table must be one of: {', '.join(_VOUCHER_TARGETS)}")
    in_use = conn.execute(
        f"SELECT COUNT(*) AS c FROM {target} WHERE voucher_type = ?", (vt,)).fetchone()["c"]
    if in_use:
        err(f"Cannot deactivate voucher_type '{vt}' for {target}: {in_use} row(s) still use it.")
    conn.execute(
        "UPDATE voucher_type_registry SET is_active = 0 WHERE voucher_type = ? AND target_table = ?",
        (vt, target))
    if conn.total_changes == 0:
        err(f"voucher_type '{vt}' is not registered for {target}")
    audit(conn, "erpclaw-setup", "update", "voucher_type_registry", f"{vt}/{target}",
          new_values={"is_active": 0})
    conn.commit()
    ok({"result": "deactivated", "voucher_type": vt, "target_table": target})


def validate_registry_completeness(conn, args):
    """Diagnostic: report type/status values used in live data that are NOT
    registered+active (i.e. would now be rejected on new writes). Read-only."""
    def _distinct(table, col, extra=""):
        try:
            return {r[0] for r in conn.execute(
                f"SELECT DISTINCT {col} FROM {table} WHERE {col} IS NOT NULL AND {col} != ''{extra}")}
        except sqlite3.OperationalError:
            return set()  # table not present on a minimal install

    def _active(table, col, target=None):
        sql = f"SELECT {col} FROM {table} WHERE is_active = 1"
        params = ()
        if target:
            sql += " AND target_table = ?"
            params = (target,)
        return {r[0] for r in conn.execute(sql, params)}

    report = {}
    # account_type
    used = _distinct("account", "account_type")
    report["account_type"] = sorted(used - _active("account_type_registry", "account_type"))
    # party_type (across the tables that carry it)
    party_used = (_distinct("gl_entry", "party_type") | _distinct("payment_entry", "party_type")
                  | _distinct("payment_ledger_entry", "party_type"))
    report["party_type"] = sorted(party_used - _active("party_type_registry", "party_type"))
    # voucher_type per target table
    for target in _VOUCHER_TARGETS:
        used = _distinct(target, "voucher_type")
        report[f"voucher_type[{target}]"] = sorted(
            used - _active("voucher_type_registry", "voucher_type", target))
    # asset status
    used = _distinct("asset", "status")
    report["asset_status"] = sorted(used - _active("asset_status_registry", "status"))

    unregistered = {k: v for k, v in report.items() if v}
    ok({"complete": not unregistered,
        "unregistered_in_use": unregistered,
        "detail": "Values listed are present in live data but not registered+active; "
                  "new writes using them would be rejected. Register via add-*-type."})


def migrate_action(conn, args):
    """Run pending foundation migrations (migrations/NNN_*.py), recording each in
    the erpclaw_schema_migration ledger. Idempotent + dialect-aware. --dry-run
    lists pending without applying. (Manages its own DB connections via db_path.)"""
    import importlib.util
    runner_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "migration_runner.py")
    spec = importlib.util.spec_from_file_location("migration_runner", runner_path)
    runner = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(runner)
    db_path = getattr(args, "db_path", None) or DEFAULT_DB_PATH
    res = runner.run_pending(db_path, dry_run=bool(getattr(args, "dry_run", False)))
    if res.get("ok") is False:
        err(f"Migration '{res['failed']}' failed: {res['error']}",
            suggestion=res.get("detail"))
    ok(res)


# ---------------------------------------------------------------------------
# Custom fields (M1 — UDF runtime admin surface; wraps erpclaw_lib.custom_fields)
# ---------------------------------------------------------------------------

_VALID_CF_TYPES = ("text", "int", "float", "date", "select", "link", "json")


def add_custom_field_action(conn, args):
    """Register a custom field definition on a core table (M1)."""
    table = (args.table or "").strip()
    field = (args.field_name or "").strip()
    if not table or not field:
        err("--table and --field-name are required")
    ftype = (args.field_type or "").strip()
    if ftype not in _VALID_CF_TYPES:
        err(f"--field-type must be one of: {', '.join(_VALID_CF_TYPES)}")
    # --options is a convenience for select/link types; stored as the lib's
    # field_options JSON. For select: comma list -> {"values": [...]}.
    field_options = None
    if args.options:
        if ftype == "select":
            field_options = json.dumps(
                {"values": [o.strip() for o in args.options.split(",") if o.strip()]})
        elif ftype == "link":
            field_options = json.dumps({"table": args.options.strip()})
        else:
            field_options = args.options  # caller-supplied raw JSON
    try:
        field_id = cf.add_custom_field(
            conn, table, field, ftype, owner_skill=args.skill_name or "erpclaw-setup",
            label=args.label, required=bool(args.required),
            default_value=args.default, field_options=field_options)
    except sqlite3.IntegrityError:
        err(f"Custom field '{field}' already exists on {table}")
    audit(conn, "erpclaw-setup", "create", "custom_field", field_id,
          new_values={"table_name": table, "field_name": field, "field_type": ftype})
    conn.commit()
    ok({"result": "registered", "custom_field_id": field_id,
        "table": table, "field_name": field, "field_type": ftype})


def list_custom_fields_action(conn, args):
    """List custom field definitions (optionally filtered by --table)."""
    t = Table("custom_field")
    q = Q.from_(t).select(t.star)
    params = ()
    if args.table:
        q = q.where(t.table_name == P())
        params = (args.table,)
    rows = conn.execute(q.orderby(t.table_name).orderby(t.field_name).get_sql(), params).fetchall()
    ok({"custom_fields": [row_to_dict(r) for r in rows], "count": len(rows)})


def remove_custom_field_action(conn, args):
    """Remove a custom field definition + its stored values.

    Hard delete (matches the shipped lib). Guarded: refuses if any stored values
    exist unless --confirm is passed, so you can't silently drop live data.
    """
    table = (args.table or "").strip()
    field = (args.field_name or "").strip()
    if not table or not field:
        err("--table and --field-name are required")
    defn = conn.execute(
        "SELECT owner_skill FROM custom_field WHERE table_name = ? AND field_name = ?",
        (table, field)).fetchone()
    if not defn:
        err(f"Custom field '{field}' not found on {table}")
    in_use = conn.execute(
        "SELECT COUNT(*) AS c FROM custom_field_value WHERE table_name = ? AND field_name = ?",
        (table, field)).fetchone()["c"]
    if in_use and not args.confirm:
        err(f"Custom field '{field}' has {in_use} stored value(s). "
            f"Pass --confirm to delete the field and its values.")
    removed = cf.remove_custom_field(conn, table, field, owner_skill=defn["owner_skill"])
    if not removed:
        err(f"Could not remove '{field}' (owner mismatch)")
    audit(conn, "erpclaw-setup", "delete", "custom_field", f"{table}/{field}",
          new_values={"deleted_values": in_use})
    conn.commit()
    ok({"result": "removed", "table": table, "field_name": field, "deleted_values": in_use})


def set_custom_field_value_action(conn, args):
    """Set a single custom field value on a row (validated)."""
    table = (args.table or "").strip()
    if not table or not args.row_id or not args.field_name:
        err("--table, --row-id and --field-name are required")
    if args.value is None:
        err("--value is required")
    values = {args.field_name: args.value}
    errors = cf.validate_custom_field_values(conn, table, values)
    if errors:
        err("; ".join(errors))
    cf.store_custom_field_values(conn, table, args.row_id, values)
    conn.commit()
    ok({"result": "stored", "table": table, "row_id": args.row_id,
        "field_name": args.field_name, "value": args.value})


def get_custom_field_values_action(conn, args):
    """Get stored custom field values for a row (all, or one --field-name)."""
    table = (args.table or "").strip()
    if not table or not args.row_id:
        err("--table and --row-id are required")
    values = cf.fetch_custom_field_values(conn, table, args.row_id)
    if args.field_name:
        values = {args.field_name: values.get(args.field_name)}
    ok({"table": table, "row_id": args.row_id, "custom_fields": values})


def set_advance_account(conn, args):
    """Configure the B1-style advance sub-account on a company (S2).

    --type customer -> company.advance_from_customer_account_id (must be a
    liability account; we owe the customer until delivery).
    --type supplier -> company.advance_to_supplier_account_id (must be an asset
    account; the supplier owes us delivery). When set, submit-payment routes the
    unallocated advance leg here instead of the AR/AP control account.
    """
    company_id = getattr(args, "company_id", None)
    acct_id = getattr(args, "account_id", None)
    kind = (getattr(args, "type", None) or "").strip().lower()
    if not company_id or not acct_id:
        err("--company-id and --account-id are required")
    if kind not in ("customer", "supplier"):
        err("--type must be 'customer' or 'supplier'")
    if not conn.execute("SELECT 1 FROM company WHERE id = ?", (company_id,)).fetchone():
        err(f"Company {company_id} not found")
    acct = conn.execute(
        "SELECT root_type, is_group FROM account WHERE id = ?", (acct_id,)).fetchone()
    if not acct:
        err(f"Account {acct_id} not found")
    if acct["is_group"]:
        err("Advance account must be a ledger (non-group) account")
    required_root = "liability" if kind == "customer" else "asset"
    if acct["root_type"] != required_root:
        err(f"Advance-from-{kind} account must be a '{required_root}' account, "
            f"got '{acct['root_type']}'")
    column = ("advance_from_customer_account_id" if kind == "customer"
              else "advance_to_supplier_account_id")
    # dialect-aware updated_at (now() -> CAST(CURRENT_TIMESTAMP AS TEXT) on SQLite, NOW()::text
    # on Postgres); column is one of two allowlisted names above.
    sql, params = dynamic_update(
        "company", {column: acct_id, "updated_at": now()}, {"id": company_id})
    conn.execute(sql, params)
    audit(conn, "erpclaw-setup", "update", "company", company_id,
          new_values={column: acct_id})
    conn.commit()
    ok({"result": "set", "company_id": company_id, "type": kind,
        "column": column, "account_id": acct_id})


ACTIONS = {
    "initialize-database": initialize_database,
    "migrate": migrate_action,
    "add-account-type": add_account_type,
    "list-account-types": list_account_types,
    "deactivate-account-type": deactivate_account_type,
    "add-voucher-type": add_voucher_type,
    "list-voucher-types": list_voucher_types,
    "deactivate-voucher-type": deactivate_voucher_type,
    "validate-registry-completeness": validate_registry_completeness,
    "add-custom-field": add_custom_field_action,
    "list-custom-fields": list_custom_fields_action,
    "remove-custom-field": remove_custom_field_action,
    "set-custom-field-value": set_custom_field_value_action,
    "get-custom-field-values": get_custom_field_values_action,
    "set-advance-account": set_advance_account,
    "set-credential": set_credential_action,
    "get-credential": get_credential_action,
    "list-credentials": list_credentials_action,
    "delete-credential": delete_credential_action,
    "migrate-credentials": migrate_credentials_action,
    "import-master-key-from-backup": import_master_key_from_backup_action,
    "setup-company": setup_company,
    "update-company": update_company,
    "get-company": get_company,
    "list-companies": list_companies,
    "add-currency": add_currency,
    "list-currencies": list_currencies,
    "add-exchange-rate": add_exchange_rate,
    "get-exchange-rate": get_exchange_rate,
    "list-exchange-rates": list_exchange_rates,
    "add-payment-terms": add_payment_terms,
    "list-payment-terms": list_payment_terms,
    "add-uom": add_uom,
    "list-uoms": list_uoms,
    "add-uom-conversion": add_uom_conversion,
    "seed-defaults": seed_defaults,
    "get-audit-log": get_audit_log,
    "get-schema-version": get_schema_version,
    "update-regional-settings": update_regional_settings,
    "backup-database": backup_database,
    "list-backups": list_backups,
    "verify-backup": verify_backup,
    "restore-database": restore_database,
    "cleanup-backups": cleanup_backups,
    "fetch-exchange-rates": fetch_exchange_rates,
    "status": status,
    "tutorial": tutorial,
    "add-user": add_user,
    "update-user": update_user,
    "list-users": list_users,
    "get-user": get_user,
    "add-role": add_role,
    "list-roles": list_roles,
    "assign-role": assign_role,
    "revoke-role": revoke_role,
    "set-password": set_password,
    "seed-permissions": seed_permissions,
    "link-telegram-user": link_telegram_user,
    "unlink-telegram-user": unlink_telegram_user,
    "check-telegram-permission": check_telegram_permission,
    "onboarding-step": onboarding_step,
}


def main():
    parser = SafeArgumentParser(description="ERPClaw Setup Skill")
    parser.add_argument("--action", required=True, choices=sorted(ACTIONS.keys()))
    parser.add_argument("--db-path", default=None,
                        help="SQLite database path (default: ~/.openclaw/erpclaw/data.sqlite)")

    # Credential management flags (set-credential / get-credential / etc.)
    parser.add_argument("--integration", default=None,
                        help="Integration name (stripe, shopify, etc.)")
    parser.add_argument("--value", default=None,
                        help="Credential value (avoid in shared shell history)")
    parser.add_argument("--from-stdin", action="store_true",
                        help="Read credential value from stdin (recommended for automation)")
    parser.add_argument("--from-env", default=None,
                        help="Read credential value from named env var")
    parser.add_argument("--dry-run", action="store_true",
                        help="Preview migrate-credentials without writing")

    # import-master-key-from-backup flags
    # (--backup-path and --passphrase reuse the backup-database flags above)
    parser.add_argument("--passphrase-from-stdin", action="store_true",
                        help="Read passphrase from stdin (recommended for automation)")
    parser.add_argument("--passphrase-from-env", default=None,
                        help="Read passphrase from named env var")

    # Company flags
    parser.add_argument("--name", default=None)
    parser.add_argument("--abbr", default=None)
    parser.add_argument("--currency", default=None)
    parser.add_argument("--country", default=None)
    parser.add_argument("--industry", default=None)
    parser.add_argument("--company-id", default=None)
    parser.add_argument("--tax-id", default=None)
    parser.add_argument("--fiscal-year-start-month", type=int, default=None)
    parser.add_argument("--default-receivable-account-id", default=None)
    parser.add_argument("--default-payable-account-id", default=None)
    parser.add_argument("--default-income-account-id", default=None)
    parser.add_argument("--default-expense-account-id", default=None)
    parser.add_argument("--default-cost-center-id", default=None)
    parser.add_argument("--default-warehouse-id", default=None)
    parser.add_argument("--default-bank-account-id", default=None)
    parser.add_argument("--default-cash-account-id", default=None)
    parser.add_argument("--round-off-account-id", default=None)
    parser.add_argument("--exchange-gain-loss-account-id", default=None)
    parser.add_argument("--perpetual-inventory", type=int, default=None)
    parser.add_argument("--enable-negative-stock", type=int, default=None)
    parser.add_argument("--accounts-frozen-till-date", default=None)
    parser.add_argument("--role-allowed-for-frozen-entries", default=None)

    # Currency flags
    parser.add_argument("--code", default=None)
    parser.add_argument("--symbol", default=None)
    parser.add_argument("--decimal-places", type=int, default=None)
    parser.add_argument("--enabled", action="store_true", default=False)
    parser.add_argument("--enabled-only", action="store_true", default=False)
    parser.add_argument("--from-currency", default=None)
    parser.add_argument("--to-currency", default=None)
    parser.add_argument("--rate", default=None)
    parser.add_argument("--effective-date", default=None)
    parser.add_argument("--source", default=None)

    # Payment terms flags
    parser.add_argument("--due-days", type=int, default=None)
    parser.add_argument("--discount-percentage", default=None)
    parser.add_argument("--discount-days", type=int, default=None)
    parser.add_argument("--description", default=None)

    # UoM flags
    parser.add_argument("--must-be-whole-number", action="store_true", default=False)
    parser.add_argument("--from-uom", default=None)
    parser.add_argument("--to-uom", default=None)
    parser.add_argument("--conversion-factor", default=None)
    parser.add_argument("--item-id", default=None)

    # Audit log flags
    parser.add_argument("--entity-type", default=None)
    parser.add_argument("--entity-id", default=None)
    parser.add_argument("--audit-action", default=None)
    parser.add_argument("--from-date", default=None)
    parser.add_argument("--to-date", default=None)
    parser.add_argument("--limit", type=int, default=None)
    parser.add_argument("--offset", type=int, default=None)

    # Schema version flags
    parser.add_argument("--module", default=None)

    # Regional settings flags
    parser.add_argument("--date-format", default=None)
    parser.add_argument("--number-format", default=None)
    parser.add_argument("--default-tax-template-id", default=None)

    # RBAC flags
    parser.add_argument("--user-id", default=None)
    parser.add_argument("--email", default=None)
    parser.add_argument("--full-name", default=None)
    parser.add_argument("--user-status", default=None)
    parser.add_argument("--role-name", default=None)
    parser.add_argument("--password", default=None,
                        help="Password for set-password action")
    parser.add_argument("--telegram-user-id", default=None,
                        help="Telegram numeric user ID for link/check actions")
    parser.add_argument("--skill", default=None,
                        help="Skill name for permission check")
    parser.add_argument("--check-action", default=None,
                        help="Action name for permission check")

    # Backup flags
    parser.add_argument("--backup-path", default=None)
    parser.add_argument("--encrypt", action="store_true", default=False,
                        help="Encrypt the backup with AES-256")
    parser.add_argument("--passphrase", default=None,
                        help="Passphrase for encrypted backup/restore")

    # Onboarding wizard flags
    parser.add_argument("--answer", default=None,
                        help="User's answer for the current onboarding step")
    parser.add_argument("--reset", action="store_true", default=False,
                        help="Reset onboarding wizard to step 1")

    # Initialize-database flags
    parser.add_argument("--force", action="store_true", default=False,
                        help="Force re-initialize: drop and recreate the database")

    # Migrate: reuses the existing --dry-run flag above (list pending without applying)

    # Type/status registry administration (M0)
    parser.add_argument("--account-type", default=None, help="Registry: account_type value")
    parser.add_argument("--voucher-type", default=None, help="Registry: voucher_type value")
    parser.add_argument("--target-table", default=None,
                        help="Registry: voucher_type target (gl_entry|stock_ledger_entry|payment_allocation)")
    parser.add_argument("--label", default=None, help="Registry: human-readable label")
    parser.add_argument("--skill-name", default=None, help="Registry: owning skill (default 'custom')")
    parser.add_argument("--include-inactive", action="store_true", default=False,
                        help="Registry list: include is_active=0 rows")

    # Advance account config (S2)
    parser.add_argument("--account-id", default=None, help="set-advance-account: the GL account id")
    parser.add_argument("--type", default=None, help="set-advance-account: 'customer' or 'supplier'")

    # Custom fields (M1 — UDF runtime). --label / --skill-name reused from above.
    parser.add_argument("--table", default=None, help="Custom field: target table (e.g. customer)")
    parser.add_argument("--field-name", default=None, help="Custom field: field name (snake_case)")
    parser.add_argument("--field-type", default=None,
                        help="Custom field: text|int|float|date|select|link|json")
    parser.add_argument("--default", default=None, help="Custom field: default value")
    parser.add_argument("--required", action="store_true", default=False,
                        help="Custom field: mark required")
    parser.add_argument("--options", default=None,
                        help="Custom field: select -> comma list; link -> target table; else raw JSON")
    parser.add_argument("--row-id", default=None, help="Custom field value: parent row id")
    # --value reused from the existing config flag above (set-custom-field-value)
    parser.add_argument("--confirm", action="store_true", default=False,
                        help="Custom field: confirm destructive removal of a field with stored values")

    args, unknown = parser.parse_known_args()
    check_unknown_args(parser, unknown)
    check_input_lengths(args)

    # initialize-database handles its own connection lifecycle
    if args.action == "initialize-database":
        try:
            initialize_database(None, args)
        except Exception as e:
            err(str(e))
        return

    # migrate runs migration scripts that open their own connections; don't hold
    # one open here (avoids SQLite writer contention during table rebuilds)
    if args.action == "migrate":
        try:
            migrate_action(None, args)
        except Exception as e:
            err(str(e))
        return

    # Connect to database
    db_path = args.db_path or DEFAULT_DB_PATH
    ensure_db_exists(db_path)
    conn = get_connection(db_path)

    try:
        ACTIONS[args.action](conn, args)
    except Exception as e:
        err(str(e))
    finally:
        conn.close()


if __name__ == "__main__":
    main()