#!/usr/bin/env python3
# PawFlow CLI

"""
Point d'entree principal de PawFlow.
Usage:
    python cli.py run <flow.json> [--input <file>] [--verbose]
    python cli.py validate <flow.json>
    python cli.py list-tasks
    python cli.py info <flow.json>
    python cli.py serve [--host] [--port] [--reload]
    python cli.py gui [--host] [--port] [--headless]
    python cli.py import <file> [-o output]
    python cli.py cluster status [--api-url]
"""

import argparse
import getpass
import json
import logging
import os
import sys
import time
from pathlib import Path

# Enregistrer toutes les tasks avant le parsing
import tasks  # noqa: F401 - declenche register_all_tasks()

from core import FlowFile, TaskFactory, FlowError
from engine.parser import FlowParser, FlowValidator
from engine.continuous_executor import ContinuousFlowExecutor
from engine.provenance import ProvenanceRepository


def cmd_run(args):
    """Executer un flow depuis un fichier JSON."""
    flow_path = Path(args.flow)
    if not flow_path.exists():
        print(f"ERREUR: Fichier introuvable: {flow_path}")
        return 1

    # Configurer le logging
    level = logging.DEBUG if args.verbose else logging.INFO
    logging.basicConfig(
        level=level,
        format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
        datefmt='%H:%M:%S'
    )
    logger = logging.getLogger('PawFlow')

    # Parser le flow
    logger.info(f"Chargement du flow: {flow_path}")
    try:
        flow = FlowParser.parse_from_file(str(flow_path))
    except Exception as e:
        print(f"ERREUR: Impossible de parser le flow: {e}")
        return 1

    logger.info(f"Flow: {flow.name} (id={flow.id})")
    logger.info(f"Tasks: {list(flow.tasks.keys())}")
    logger.info(f"Relations: {len(flow.relations)}")

    # Valider
    errors = FlowValidator.validate(flow, strict=False)
    if errors:
        for err in errors:
            logger.warning(f"Validation: {err}")

    # Creer les FlowFiles d'entree
    input_flowfiles = []
    if args.input:
        for input_path in args.input:
            p = Path(input_path)
            if not p.exists():
                print(f"ERREUR: Fichier d'entree introuvable: {p}")
                return 1
            content = p.read_bytes()
            ff = FlowFile(
                content=content,
                attributes={
                    'filename': p.name,
                    'path': str(p.parent),
                    'absolute.path': str(p.resolve()),
                    'fileSize': str(len(content)),
                }
            )
            input_flowfiles.append(ff)
            logger.info(f"Input: {p.name} ({len(content)} bytes)")
    else:
        # FlowFile vide par defaut
        input_flowfiles = [FlowFile(content=b'', attributes={'filename': 'stdin'})]
        logger.info("Pas d'input specifie, utilisation d'un FlowFile vide")

    # Provenance
    repo = ProvenanceRepository() if args.provenance else None

    # Parse --param overrides
    param_overrides = {}
    if args.param:
        for p in args.param:
            if '=' not in p:
                print(f"ERREUR: Format invalide pour --param: '{p}' (attendu KEY=VALUE)")
                return 1
            k, v = p.split('=', 1)
            param_overrides[k.strip()] = v.strip()
        logger.info(f"Parameter overrides: {param_overrides}")

    # Execute (batch mode via ContinuousFlowExecutor)
    logger.info("--- Debut de l'execution ---")
    start = time.time()
    result = ContinuousFlowExecutor.run_batch(
        flow,
        input_flowfiles=input_flowfiles,
        parameters=param_overrides if param_overrides else None,
        max_workers=args.workers,
        max_retries=args.retries,
        timeout=args.timeout or 300,
        provenance=repo,
    )
    elapsed = time.time() - start

    # Afficher le resultat
    print()
    if result.success:
        print(f"SUCCES - {flow.name}")
    else:
        print(f"ECHEC - {flow.name}")

    print(f"  Duree: {elapsed:.3f}s")
    print(f"  FlowFiles en sortie: {len(result.output_flowfiles)}")

    if result.errors:
        print(f"  Erreurs:")
        for err in result.errors:
            print(f"    - {err.get('error', 'unknown')}")

    # Stats par task
    if result.task_statistics:
        print(f"  Statistiques par tache:")
        for tid, stats in result.task_statistics.items():
            print(f"    {tid} ({stats.task_type}): "
                  f"{stats.success_count} ok, {stats.error_count} err, "
                  f"{stats.avg_duration_ms:.1f}ms avg")

    # Provenance
    if repo and repo.size() > 0:
        prov = repo.to_dict()
        print(f"  Provenance: {prov['total_events']} evenements")
        for etype, count in prov['events_by_type'].items():
            print(f"    {etype}: {count}")

    # Output
    if args.output_dir:
        out_dir = Path(args.output_dir)
        out_dir.mkdir(parents=True, exist_ok=True)
        for i, ff in enumerate(result.output_flowfiles):
            fname = ff.get_attribute('filename', f'output_{i}.dat')
            out_path = out_dir / fname
            out_path.write_bytes(ff.content)
            logger.info(f"Output: {out_path}")
        print(f"  Fichiers ecrits dans: {out_dir}")

    return 0 if result.success else 1


def cmd_validate(args):
    """Valider un flow JSON."""
    flow_path = Path(args.flow)
    if not flow_path.exists():
        print(f"ERREUR: Fichier introuvable: {flow_path}")
        return 1

    try:
        flow = FlowParser.parse_from_file(str(flow_path))
        errors = FlowValidator.validate(flow, strict=False)
    except Exception as e:
        print(f"ERREUR: {e}")
        return 1

    if errors:
        print(f"VALIDATION ECHOUEE pour {flow.name}:")
        for err in errors:
            print(f"  - {err}")
        return 1
    else:
        print(f"VALIDE: {flow.name} ({len(flow.tasks)} tasks, {len(flow.relations)} relations)")
        return 0


def cmd_list_tasks(args):
    """Lister toutes les tasks disponibles."""
    print("Tasks disponibles:")
    print(f"{'TYPE':<25} {'NOM':<30} {'VERSION'}")
    print("-" * 65)
    for task_type in sorted(TaskFactory.list_types()):
        cls = TaskFactory.get(task_type)
        print(f"{cls.TYPE:<25} {cls.NAME:<30} {cls.VERSION}")


def cmd_info(args):
    """Afficher les infos d'un flow."""
    flow_path = Path(args.flow)
    if not flow_path.exists():
        print(f"ERREUR: Fichier introuvable: {flow_path}")
        return 1

    try:
        flow = FlowParser.parse_from_file(str(flow_path))
    except Exception as e:
        print(f"ERREUR: {e}")
        return 1

    print(f"Flow: {flow.name}")
    print(f"  ID: {flow.id}")
    print(f"  Version: {flow.version}")
    print(f"  Description: {flow.description}")
    print(f"  Tasks ({len(flow.tasks)}):")
    for tid, task in flow.tasks.items():
        print(f"    - {tid} (type={task.TYPE})")
    print(f"  Relations ({len(flow.relations)}):")
    for rel in flow.relations:
        print(f"    - {rel['from']} -> {rel['to']}")
    return 0



def _log_startup_urls(logger, host: str, port: int, install_complete: bool) -> None:
    """Log only the URL that is valid for the current startup phase."""
    base_url = f"https://{host}:{port}"
    if install_complete:
        logger.info("  Chat:    %s/chat", base_url)
    else:
        logger.info("  Install: %s/install", base_url)


def cmd_start(args):
    """Start PawFlow server.

    The PawFlow server (chat, agents, HTTP listener) runs in the main process.
    Admin GUI is served natively at /admin via the pawflow-admin flow.
    """
    import signal

    # Configure logging
    from core.server_logging import configure_server_logging
    configure_server_logging(logging.INFO)
    logger = logging.getLogger("pawflow")

    try:
        from core.cli_workspace_mounts import set_workspace_mount_mode
        _mount_arg = getattr(args, "workspace_mount", None)
        if _mount_arg is not None:
            _mount_mode = set_workspace_mount_mode(_mount_arg)
            logger.info("CLI workspace fallback mount mode: %s", _mount_mode)
    except Exception as _wm_err:
        logger.warning("CLI workspace mount mode setup failed: %s", _wm_err)

    # 1. Register tasks and restore flows in the main process
    # Cleanup orphan Docker containers from previous server run
    from core.docker_utils import get_server_id, kill_containers
    _srv_id = get_server_id()
    _killed = kill_containers(_srv_id)
    if _killed:
        logger.info("Cleaned up %d orphan Docker container(s) from previous run", _killed)

    # Initialise the capability-auth store before any HTTP/WS handler can
    # reach it. The JSON store persists across restarts so VNC/terminal/
    # code-server/port-forward sessions a user opened before the restart
    # still match their capability tokens after.
    from core.capability_auth import init_db as _init_capabilities
    from core.paths import CAPABILITIES_FILE
    _init_capabilities(CAPABILITIES_FILE)

    logger.info("Registering tasks...")
    from tasks import register_all_tasks
    register_all_tasks()

    try:
        from core.install_bootstrap import ensure_install_bootstrap, is_install_complete
        ensure_install_bootstrap(port=int(args.port))
    except Exception as _ib_err:
        logger.error("Install bootstrap setup failed: %s", _ib_err, exc_info=True)
        raise

    logger.info("Restoring deployed flows...")
    from core.executor_registry import ExecutorRegistry
    er = ExecutorRegistry.get_instance()
    er.restore_from_disk()
    n = er.count()
    logger.info(f"PawFlow server ready — {n} flow(s) restored")
    _log_startup_urls(logger, args.host, int(args.port), is_install_complete())

    # Startup security report. In production mode
    # (PAWFLOW_ENV=production / PAWFLOW_PUBLIC_MODE=true) this raises
    # SystemExit if a critical setting is unsafe (weak default key,
    # fail-open approval, ...). Always logs the snapshot.
    try:
        from core.security_report import build_report, enforce
        enforce(build_report())
    except SystemExit:
        raise
    except Exception as _se:
        logger.warning("security report failed: %s", _se, exc_info=True)

    # Boot recovery: scan every conv/{agent}/pending.jsonl and wake agents
    # that had undelivered messages when the previous process died. Without
    # this, a message enqueued just before a crash would sit on disk with
    # nobody polling for it.
    try:
        from core.pending_queue import PendingQueue
        from tasks.ai.agent_loop import AgentLoopTask
        _recovered = PendingQueue.all_nonempty()
        if _recovered:
            logger.info(f"[boot-recovery] {len(_recovered)} agent(s) have "
                        f"pending messages — scheduling wake")
            for conv_id, agent, count in _recovered:
                logger.info(f"[boot-recovery] waking {conv_id[:8]}/{agent or '_shared'} "
                            f"({count} pending)")
                AgentLoopTask.wake_agent(
                    conv_id, agent,
                    reason=f"[boot-recovery] {count} pending msg(s)",
                    delay=2.0,
                )
    except Exception as _re:
        logger.warning(f"[boot-recovery] scan failed: {_re}")

    # 2. Keep main thread alive, handle Ctrl+C gracefully
    _shutting_down = False

    def _shutdown(sig, frame):
        nonlocal _shutting_down
        if _shutting_down:
            # Second Ctrl+C → force kill immediately
            logger.info("Force kill.")
            _kill_spawned_docker_containers()
            os._exit(1)
        _shutting_down = True
        logger.info("Shutting down...")

        # Stop executors in a thread with timeout
        import threading
        def _stop_all():
            try:
                reg = ExecutorRegistry.get_instance()
                for eid in list(reg._executors.keys()):
                    try:
                        reg._executors[eid].stop()
                    except Exception:
                        pass  # nosec B110
            except Exception:
                pass  # nosec B110
        t = threading.Thread(target=_stop_all, daemon=True)
        t.start()
        t.join(timeout=3)
        # Drain the ConversationWriter FIFO BEFORE os._exit. The writer
        # runs on a daemon thread; os._exit kills it instantly, and any
        # message still in the queue is gone. Executors are stopped first
        # so no new enqueues can race the drain.
        try:
            from core.conversation_writer import ConversationWriter
            if not ConversationWriter.shutdown_all(wait_timeout=20.0):
                logger.error(
                    "ConversationWriter drain INCOMPLETE - some messages "
                    "were not persisted before shutdown")
        except Exception as _cw_err:
            logger.error("ConversationWriter drain failed: %s", _cw_err,
                         exc_info=True)
        # Nothing we spawned should outlive us. Reap provider pools and
        # server-side provider login containers before exiting.
        _kill_spawned_docker_containers()
        os._exit(0)


    def _kill_spawned_docker_containers():
        """Hard-kill all containers this PawFlow server spawned.

        Covers provider pool containers (`pf-*-pool-*`), server-side
        relay containers (`pawflow-relay-srv-*`, `pawflow-relay-min-*`),
        and login containers (`pawflow-*-login-*`) for Claude Code,
        Codex, and Gemini. This is the normal `docker stop pawflow-server`
        cleanup path.
        """
        for _pool_mod, _pool_cls in (
            ("core.claude_code_pool", "ClaudeCodePool"),
            ("core.codex_pool", "CodexPool"),
            ("core.gemini_pool", "GeminiPool"),
        ):
            try:
                import importlib as _importlib
                getattr(_importlib.import_module(_pool_mod), _pool_cls).instance().shutdown()
            except Exception:
                pass  # nosec B110
        try:
            import subprocess as _sp  # nosec B404
            from core.docker_utils import docker_cmd
            for _prefix in (
                "pf-cc-pool-",
                "pf-codex-pool-",
                "pf-gemini-pool-",
                "pawflow-relay-srv-",
                "pawflow-relay-min-",
                "pawflow-claude-login-",
                "pawflow-codex-login-",
                "pawflow-gemini-login-",
            ):
                try:
                    _r = _sp.run(  # nosec B603
                        docker_cmd() + ["ps", "-a", "-q",
                                         "--filter", f"name={_prefix}"],
                        capture_output=True, text=True, timeout=5)
                    _ids = [i for i in (_r.stdout or "").split() if i]
                    if _ids:
                        _sp.run(docker_cmd() + ["rm", "-f"] + _ids,  # nosec B603
                                capture_output=True, timeout=10)
                        logger.info("Reaped %d orphan container(s) matching %s*",
                                    len(_ids), _prefix)
                except Exception:
                    pass  # nosec B110
        except Exception:
            pass  # nosec B110

    signal.signal(signal.SIGINT, _shutdown)
    signal.signal(signal.SIGTERM, _shutdown)
    # atexit belt-and-suspenders: reap spawned containers even on abnormal
    # exit (uncaught exception, sys.exit, etc.). Signal-driven _shutdown
    # already covers Ctrl-C; this covers everything else.
    import atexit as _atexit
    _atexit.register(_kill_spawned_docker_containers)

    # Guardian thread: force-exit if the shutdown handler hangs.
    # Budget must cover: executor stop (3s) + ConversationWriter drain
    # (up to 20s) + docker reap (~5s), with slack. Force-exit BEFORE the
    # drain finishes would re-introduce the message-loss bug we are
    # fixing here, so the guardian sits well past the drain budget.
    def _force_exit_guardian():
        import time as _t
        while not _shutting_down:
            _t.sleep(0.5)
        _t.sleep(45)
        logger.warning("Shutdown timeout - force exit")
        os._exit(1)
    import threading as _th_guard
    _guardian = _th_guard.Thread(target=_force_exit_guardian, daemon=True)
    _guardian.start()

    # On Windows: Ctrl-C is eaten by wsl.exe subprocess.
    # Use a dedicated thread that reads stdin for Ctrl-C detection.
    if os.name == "nt":
        def _ctrl_c_watcher():
            """Watch for Ctrl-C via msvcrt on Windows."""
            import msvcrt
            while not _shutting_down:
                if msvcrt.kbhit():
                    ch = msvcrt.getch()
                    if ch == b'\x03':  # Ctrl-C
                        logger.info("Ctrl-C detected via kbhit")
                        _shutdown(None, None)
                        return
                time.sleep(0.1)
        _watcher = _th_guard.Thread(target=_ctrl_c_watcher, daemon=True,
                                     name="ctrl-c-watcher")
        _watcher.start()

    try:
        while not _shutting_down:
            time.sleep(0.5)
    except KeyboardInterrupt:
        pass
    if not _shutting_down:
        _shutdown(None, None)


def cmd_import_flow(args):
    """Import a NiFi flow."""
    path = args.path
    from engine.nifi_converter import NiFiConverter
    with open(path) as f:
        content = f.read()
    converter = NiFiConverter()
    result = converter.convert(content)

    output_path = args.output or "imported_flow.json"
    Path(output_path).parent.mkdir(parents=True, exist_ok=True)
    with open(output_path, 'w') as f:
        json.dump(result.flow, f, indent=2)

    print(f"Imported flow: {output_path}")
    if result.warnings:
        print(f"Warnings ({len(result.warnings)}):")
        for w in result.warnings[:5]:
            print(f"  - {w}")
    if result.subflows:
        for sf in result.subflows:
            sf_path = f"{sf.get('id', 'subflow')}.json"
            with open(sf_path, 'w') as f_out:
                json.dump(sf, f_out, indent=2)
            print(f"  Subflow: {sf_path}")
    return 0


def cmd_triggers(args):
    """Manage event triggers."""
    from engine.triggers import TriggerManager, TriggerType

    tm = TriggerManager()
    from core.paths import TRIGGERS_FILE
    config_path = str(TRIGGERS_FILE)
    tm.load_triggers(config_path)

    if args.action == "list":
        triggers = tm.list_triggers()
        if not triggers:
            print("No triggers configured.")
            return 0
        print(f"{'ID':<20} {'TYPE':<15} {'STATE':<10} {'FIRES':<8} {'FLOW'}")
        print("-" * 75)
        for t in triggers:
            print(f"{t['trigger_id']:<20} {t['type']:<15} {t['state']:<10} "
                  f"{t['fire_count']:<8} {t['flow_path']}")

    elif args.action == "create":
        if not args.trigger_id or not args.trigger_type or not args.flow_path:
            print("ERREUR: --trigger-id, --trigger-type, and --flow-path are required.")
            return 1
        config = {}
        if args.config:
            try:
                config = json.loads(args.config)
            except json.JSONDecodeError as e:
                print(f"ERREUR: Invalid JSON for --config: {e}")
                return 1
        try:
            result = tm.create_trigger(
                trigger_id=args.trigger_id,
                trigger_type=TriggerType(args.trigger_type),
                flow_path=args.flow_path,
                name=args.name or args.trigger_id,
                config=config,
                enabled=True,
            )
            tm.save_triggers(config_path)
            print(f"Created trigger: {args.trigger_id} ({result['state']})")
        except ValueError as e:
            print(f"ERREUR: {e}")
            return 1

    elif args.action == "start":
        if not args.trigger_id:
            print("ERREUR: --trigger-id is required.")
            return 1
        try:
            result = tm.start_trigger(args.trigger_id)
            print(f"Started: {args.trigger_id} ({result['state']})")
        except ValueError as e:
            print(f"ERREUR: {e}")
            return 1

    elif args.action == "stop":
        if not args.trigger_id:
            print("ERREUR: --trigger-id is required.")
            return 1
        try:
            result = tm.stop_trigger(args.trigger_id)
            tm.save_triggers(config_path)
            print(f"Stopped: {args.trigger_id}")
        except ValueError as e:
            print(f"ERREUR: {e}")
            return 1

    elif args.action == "delete":
        if not args.trigger_id:
            print("ERREUR: --trigger-id is required.")
            return 1
        removed = tm.delete_trigger(args.trigger_id)
        if removed:
            tm.save_triggers(config_path)
            print(f"Deleted: {args.trigger_id}")
        else:
            print(f"Trigger '{args.trigger_id}' not found.")
            return 1

    elif args.action == "history":
        history = tm.get_history(
            trigger_id=args.trigger_id if args.trigger_id else None,
            limit=50,
        )
        if not history:
            print("No trigger history.")
            return 0
        for h in history:
            from datetime import datetime
            ts = datetime.fromtimestamp(h["timestamp"]).strftime("%Y-%m-%d %H:%M:%S")
            status = "OK" if h["flow_executed"] else "FAIL"
            event = h["event_data"].get("event", "?")
            error = f" | {h['error']}" if h["error"] else ""
            print(f"  [{ts}] {h['trigger_id']} [{status}] {event}{error}")

    elif args.action == "run":
        # Keep running with triggers active (like scheduler start)
        if not tm.list_triggers():
            print("No triggers to run. Create some first.")
            return 0
        print(f"Running {len(tm.list_triggers())} trigger(s). Press Ctrl+C to stop.")
        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            tm.stop_all()
            print("\nTriggers stopped.")

    return 0


def cmd_cluster(args):
    """Cluster management."""
    if args.action == "status":
        import urllib.request
        url = args.api_url or "http://localhost:8000"
        try:
            with urllib.request.urlopen(f"{url}/api/v1/system/cluster/status") as resp:  # nosec B310
                data = json.loads(resp.read())
                if data.get("cluster_enabled"):
                    status = data["status"]
                    print(f"Cluster: {status.get('total_instances', 0)} instances")
                    print(f"Role: {status.get('role', 'unknown')}")
                    print(f"Coordinator: {status.get('coordinator_host', 'none')}")
                    for inst in status.get("instances", []):
                        marker = " *" if inst.get("id") == status.get("instance_id") else ""
                        print(f"  {inst['id']} [{inst['role']}] {inst['host']}{marker}")
                else:
                    print("Cluster mode not enabled.")
        except Exception as e:
            print(f"Cannot reach API at {url}: {e}")
    return 0


def cmd_re_embed(args):
    """Re-embed all memories for a user with vector embeddings."""
    user_id = args.user_id
    provider = args.provider or "auto"
    api_key = args.api_key or os.environ.get("OPENAI_API_KEY", "")

    logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')

    from core.embeddings import EmbeddingProvider
    from core.memory_store import MemoryStore

    store = MemoryStore.instance()
    ep = EmbeddingProvider.instance()

    count = store.count(user_id)
    if count == 0:
        print(f"No memories found for user '{user_id}'.")
        return 0

    print(f"Re-embedding {count} memories for user '{user_id}' (provider: {provider})...")

    def embed_fn(text: str):
        results = ep.embed([text], provider=provider, api_key=api_key)
        return results[0] if results else []

    embedded = store.re_embed_all(user_id, embed_fn)
    print(f"Done. {embedded}/{count} memories re-embedded.")
    return 0


def _read_admin_password(args) -> str:
    password = args.password or ""
    if args.password_env:
        password = os.environ.get(args.password_env, "")
    if not password:
        password = getpass.getpass("Admin password: ")
        confirm = getpass.getpass("Confirm admin password: ")
        if password != confirm:
            print("ERROR: passwords do not match")
            return ""
    if not password:
        print("ERROR: password is required")
        return ""
    return password


def cmd_admin_user(args):
    """Create or repair a local PawFlow admin account."""
    from core.security import SecurityManager, Role

    username = (args.username or "").strip()
    if not username:
        print("ERROR: username is required")
        return 1

    password = _read_admin_password(args)
    if not password:
        return 1

    sm = SecurityManager.get_instance()
    if args.action == "create":
        existing = sm.get_user(username)
        if existing:
            sm.update_user(
                username,
                role=Role.ADMIN,
                password=password,
                enabled=True,
                email=args.email or None,
                display_name=args.display_name or None,
            )
            print(f"Admin user '{username}' updated and enabled.")
            return 0
        sm.create_user(
            username,
            password,
            Role.ADMIN,
            email=args.email or "",
            display_name=args.display_name or username,
        )
        print(f"Admin user '{username}' created.")
        return 0

    if args.action == "reset-password":
        user = sm.get_user(username)
        if not user:
            print(f"ERROR: user '{username}' does not exist")
            return 1
        sm.update_user(username, role=Role.ADMIN, password=password, enabled=True)
        print(f"Admin user '{username}' password reset and account enabled.")
        return 0

    print(f"ERROR: unsupported admin-user action '{args.action}'")
    return 1


def main():
    from core import __version__ as _pf_version
    parser = argparse.ArgumentParser(
        prog='pawflow',
        description='PawFlow - Self-hosted AI agent orchestration platform'
    )
    parser.add_argument('--version', action='version', version=f'pawflow {_pf_version}')
    subparsers = parser.add_subparsers(dest='command', help='Command')

    # run
    run_parser = subparsers.add_parser('run', help='Run a flow')
    run_parser.add_argument('flow', help='Flow JSON file')
    run_parser.add_argument('--input', '-i', nargs='+', help='Input files')
    run_parser.add_argument('--output-dir', '-o', help='Output directory')
    run_parser.add_argument('--verbose', '-v', action='store_true', help='Verbose mode')
    run_parser.add_argument('--provenance', '-p', action='store_true', help='Enable provenance tracking')
    run_parser.add_argument('--workers', '-w', type=int, default=4, help='Parallel workers (default: 4)')
    run_parser.add_argument('--retries', '-r', type=int, default=3, help='Max retries (default: 3)')
    run_parser.add_argument('--timeout', '-t', type=int, default=300, help='Timeout in seconds (default: 300)')
    run_parser.add_argument('--param', action='append', metavar='KEY=VALUE',
                            help='Override a flow parameter (repeatable)')

    # validate
    val_parser = subparsers.add_parser('validate', help='Validate a flow')
    val_parser.add_argument('flow', help='Flow JSON file')

    # list-tasks
    subparsers.add_parser('list-tasks', help='List available tasks')

    # info
    info_parser = subparsers.add_parser('info', help='Show flow info')
    info_parser.add_argument('flow', help='Flow JSON file')

    # start
    start_parser = subparsers.add_parser('start', help='Start PawFlow server')
    start_parser.add_argument('--host', default='localhost', help='Host (default: localhost)')
    start_parser.add_argument('--port', type=int, required=True, help='Port selected for this runtime')
    start_parser.add_argument('--workspace-mount', choices=['off', 'ro', 'rw'], default=None,
                              help='Mount linked relay workspaces into CLI provider containers: off, ro, or rw. Overrides PAWFLOW_CLI_WORKSPACE_MOUNT.')



    # import
    import_parser = subparsers.add_parser('import', help='Import a NiFi flow or .pfp plugin')
    import_parser.add_argument('path', help='File to import (.pfp, .xml, or .json)')
    import_parser.add_argument('-o', '--output', help='Output flow JSON path')

    # triggers
    triggers_parser = subparsers.add_parser('triggers', help='Manage event triggers')
    triggers_parser.add_argument('action',
                                 choices=['list', 'create', 'start', 'stop', 'delete', 'history', 'run'],
                                 help='Trigger action')
    triggers_parser.add_argument('--trigger-id', dest='trigger_id', help='Trigger ID')
    triggers_parser.add_argument('--trigger-type', dest='trigger_type',
                                 choices=['file_watcher', 'webhook', 'event', 'polling'],
                                 help='Trigger type (for create)')
    triggers_parser.add_argument('--flow-path', dest='flow_path', help='Flow JSON path (for create)')
    triggers_parser.add_argument('--name', help='Trigger name (for create)')
    triggers_parser.add_argument('--config', help='Trigger config as JSON string (for create)')

    # cluster
    cluster_parser = subparsers.add_parser('cluster', help='Cluster management')
    cluster_parser.add_argument('action', choices=['status'], help='Cluster action')
    cluster_parser.add_argument('--api-url', dest='api_url', help='API URL (default: http://localhost:8000)')

    # re-embed-memories
    reembed_parser = subparsers.add_parser('re-embed-memories',
                                            help='Re-embed all memories with vector embeddings')
    reembed_parser.add_argument('--user-id', dest='user_id', required=True, help='User ID')
    reembed_parser.add_argument('--provider', choices=['openai', 'local', 'auto'],
                                default='auto', help='Embedding provider (default: auto)')
    reembed_parser.add_argument('--api-key', dest='api_key', help='OpenAI API key (optional, uses env var)')

    # admin-user
    admin_user_parser = subparsers.add_parser(
        'admin-user',
        help='Create or repair a local admin account for recovery'
    )
    admin_user_parser.add_argument('action', choices=['create', 'reset-password'],
                                   help='Create/repair an admin, or reset an existing user password')
    admin_user_parser.add_argument('--username', required=True, help='PawFlow username')
    admin_user_parser.add_argument('--password', help='Password value. Prefer --password-env or interactive prompt.')
    admin_user_parser.add_argument('--password-env', dest='password_env',
                                   help='Environment variable containing the password')
    admin_user_parser.add_argument('--email', default='', help='Optional user email for create/update')
    admin_user_parser.add_argument('--display-name', dest='display_name', default='',
                                   help='Optional display name for create/update')

    args = parser.parse_args()

    if args.command == 'run':
        sys.exit(cmd_run(args))
    elif args.command == 'validate':
        sys.exit(cmd_validate(args))
    elif args.command == 'list-tasks':
        cmd_list_tasks(args)
    elif args.command == 'info':
        sys.exit(cmd_info(args))
    elif args.command == 'start':
        sys.exit(cmd_start(args))
    elif args.command == 'import':
        sys.exit(cmd_import_flow(args))
    elif args.command == 'triggers':
        sys.exit(cmd_triggers(args))
    elif args.command == 'cluster':
        sys.exit(cmd_cluster(args))
    elif args.command == 're-embed-memories':
        sys.exit(cmd_re_embed(args))
    elif args.command == 'admin-user':
        sys.exit(cmd_admin_user(args))
    else:
        parser.print_help()


if __name__ == '__main__':
    # Windows + Python 3.14: Playwright/Patchright needs ProactorEventLoopPolicy
    # for subprocess support.  Without this, scrapling cleanup crashes on exit.
    import sys, warnings
    if sys.platform == "win32":
        import asyncio
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", DeprecationWarning)
            asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
    main()
