mod config;
mod playbook_runner;

use anyhow::{bail, Context as AnyhowContext, Result};
use base64::prelude::*;
use clap::{Parser, Subcommand};
use config::{Config, Context};
use crossterm::{
    event::{self, DisableMouseCapture, EnableMouseCapture, Event, KeyCode},
    execute,
    terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen},
};
use rand::{rngs::OsRng, RngCore};
use ratatui::{
    backend::{Backend, CrosstermBackend},
    layout::{Constraint, Direction, Layout},
    widgets::{Block, Borders, List, ListItem, ListState, Paragraph},
    Frame, Terminal,
};
use reqwest::{
    header::{HeaderMap, HeaderValue, AUTHORIZATION},
    Client,
};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use std::collections::HashMap;
use std::fs;
use std::io;
use std::path::PathBuf;
use std::process::Command;
use std::time::{Duration, Instant};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
use url::Url;

#[derive(Parser)]
#[command(name = env!("CARGO_BIN_NAME"))]
#[command(version, about = "NoETL Command Line Tool", long_about = None)]
struct Cli {
    #[command(subcommand)]
    command: Option<Commands>,

    /// Interactive mode (TUI)
    #[arg(short, long)]
    interactive: bool,

    /// NoETL server host
    #[arg(long)]
    host: Option<String>,

    /// NoETL server port
    #[arg(short, long)]
    port: Option<u16>,

    /// NoETL server URL (overrides host and port)
    #[arg(long)]
    server_url: Option<String>,

    /// Route API requests via Gateway protected proxy (/noetl/*)
    #[arg(long)]
    gateway: bool,

    /// Gateway session token (or set NOETL_SESSION_TOKEN env var)
    #[arg(long)]
    session_token: Option<String>,

    /// Use this named context for one command, without changing the
    /// current context.  Matches the UX of ``kubectl --context`` and
    /// ``gcloud --account``.  When set, the named context's
    /// ``server_url``, ``runtime``, Auth0 settings, and cached
    /// session token are used in place of the current context.
    ///
    /// Errors if the named context doesn't exist (use
    /// ``noetl context list`` to see configured contexts).
    ///
    /// Examples:
    ///     noetl --context gke-pf catalog list Playbook
    ///     noetl --context gke-prod register credential -f cred.json
    ///     noetl --context smoke-test exec ./playbooks/foo.yaml
    #[arg(long, value_name = "NAME", global = true)]
    context: Option<String>,
}

#[derive(Subcommand)]
enum Commands {
    /// Execute a playbook (unified command for local and distributed execution)
    ///
    /// The <ref> can be:
    ///   - A file path: ./playbooks/foo.yaml, automation/deploy.yaml
    ///   - A catalog reference: catalog://amadeus_ai_api@2.0
    ///   - A database ID: pbk_01J... (playbook ID from catalog)
    ///   - A catalog path: workflows/etl-pipeline (requires --runtime distributed)
    ///
    /// Runtime modes:
    ///   - local: Execute directly using Rust interpreter (no server required)
    ///   - distributed: Execute via NoETL server-worker architecture
    ///   - auto (default): Choose based on ref type and playbook executor.profile
    ///
    /// Examples:
    ///     noetl exec ./playbooks/http_test.yaml                    # auto runtime
    ///     noetl exec ./playbooks/http_test.yaml -r local           # force local
    ///     noetl exec catalog://amadeus_ai_api@2.0 -r distributed   # catalog + distributed
    ///     noetl exec my-playbook --runtime distributed             # catalog path
    ///     noetl exec ./foo.yaml --set key=value --verbose          # with variables
    ///     noetl exec pbk_01J... -r distributed                     # by db ID
    #[command(verbatim_doc_comment, alias = "run")]
    Exec {
        /// Playbook reference: file path, catalog://name@version, db ID, or catalog path
        #[arg(value_name = "REF")]
        reference: String,

        /// Runtime mode: local (Rust interpreter), distributed (server-worker), auto
        #[arg(short = 'r', long, default_value = "auto")]
        runtime: String,

        /// Target step to start execution from (local runtime only)
        #[arg(short = 't', long)]
        target: Option<String>,

        /// Set variables (format: key=value), can be repeated
        #[arg(long = "set", value_name = "KEY=VALUE")]
        variables: Vec<String>,

        /// Payload/workload as JSON string (merges with playbook workload)
        #[arg(long = "payload", value_name = "JSON", alias = "workload")]
        payload: Option<String>,

        /// Path to JSON file with input parameters
        #[arg(short, long)]
        input: Option<PathBuf>,

        /// Catalog version (for catalog:// refs without @version)
        #[arg(short = 'V', long)]
        version: Option<String>,

        /// Server endpoint for distributed runtime (default: from config)
        #[arg(long)]
        endpoint: Option<String>,

        /// Enable verbose output
        #[arg(short, long)]
        verbose: bool,

        /// Dry-run mode: validate and show plan without executing
        #[arg(long)]
        dry_run: bool,

        /// Emit only JSON response (distributed runtime)
        #[arg(short, long)]
        json: bool,
    },
    /// Legacy run command (deprecated, use 'exec')
    #[command(name = "run-legacy", hide = true)]
    RunLegacy {
        /// Playbook reference: file path, catalog://name@version, db ID, or catalog path
        #[arg(value_name = "REF")]
        reference: Option<String>,

        /// Runtime mode: local (Rust interpreter), distributed (server-worker), auto
        #[arg(short = 'r', long, default_value = "auto")]
        runtime: String,

        /// Additional arguments (target step or variables)
        #[arg(trailing_var_arg = true)]
        args: Vec<String>,

        /// Set variables (format: key=value)
        #[arg(long = "set", value_name = "KEY=VALUE")]
        variables: Vec<String>,

        /// Payload/workload as JSON string
        #[arg(long = "payload", value_name = "JSON", alias = "workload")]
        payload: Option<String>,

        /// Deep merge payload with playbook workload
        #[arg(long)]
        merge: bool,

        /// Enable verbose output
        #[arg(short, long)]
        verbose: bool,
    },
    /// Context management
    Context {
        #[command(subcommand)]
        command: ContextCommand,
    },
    /// Interactive CLI console (REPL) with connection-aware prompt
    ///
    /// Examples:
    ///     noetl console
    ///     noetl --gateway console
    ///     noetl --server-url https://gateway.example.com console
    #[command(verbatim_doc_comment)]
    Console,
    /// Run Codex CLI under noetl with passthrough behavior
    ///
    /// Examples:
    ///     noetl codex
    ///     noetl codex exec "summarize this repo"
    ///     noetl codex doctor
    #[command(verbatim_doc_comment, trailing_var_arg = true, allow_hyphen_values = true)]
    Codex {
        /// Arguments forwarded to upstream codex CLI
        args: Vec<String>,
    },
    /// Start NoETL AI mode (M1 scaffold)
    ///
    /// Examples:
    ///     noetl ai
    ///     noetl ai exec "inspect gateway status"
    #[command(verbatim_doc_comment, trailing_var_arg = true, allow_hyphen_values = true)]
    Ai {
        /// Arguments forwarded to upstream codex CLI
        args: Vec<String>,
    },
    /// Gateway authentication management
    Auth {
        #[command(subcommand)]
        command: AuthCommand,
    },
    /// Register resources
    Register {
        #[command(subcommand)]
        resource: RegisterResource,
    },
    /// Fetch execution status (legacy command, use 'execute status' instead)
    /// Examples:
    ///     noetl status 12345
    ///     noetl --host=localhost --port=8082 status 12345 --json
    #[command(verbatim_doc_comment)]
    Status {
        /// Execution ID
        execution_id: String,

        /// Emit only the JSON response
        #[arg(short, long)]
        json: bool,
    },
    /// Cancel a running execution
    /// Examples:
    ///     noetl cancel 12345
    ///     noetl cancel 12345 --reason "Detected infinite loop"
    ///     noetl cancel 12345 --cascade
    ///     noetl --host=localhost --port=8082 cancel 12345 --json
    #[command(verbatim_doc_comment)]
    Cancel {
        /// Execution ID to cancel
        execution_id: String,

        /// Reason for cancellation (optional)
        #[arg(short, long)]
        reason: Option<String>,

        /// Also cancel child executions (sub-playbooks)
        #[arg(long)]
        cascade: bool,

        /// Emit only the JSON response
        #[arg(short, long)]
        json: bool,
    },
    /// List resources in catalog (legacy command, use 'catalog list' instead)
    /// Examples:
    ///     noetl list Playbook
    ///     noetl list Credential --json
    ///     noetl --host=localhost --port=8082 list Playbook
    #[command(verbatim_doc_comment)]
    List {
        /// Resource type (e.g., Playbook, Credential)
        resource_type: String,

        /// Emit only the JSON response
        #[arg(short, long)]
        json: bool,
    },
    /// Catalog management
    Catalog {
        #[command(subcommand)]
        command: CatalogCommand,
    },
    /// Legacy execution management (use 'exec' instead)
    #[command(hide = true)]
    Execute {
        #[command(subcommand)]
        command: ExecuteCommand,
    },
    /// Get resource details
    Get {
        #[command(subcommand)]
        resource: GetResource,
    },
    /// Execute SQL query via NoETL Postgres API
    /// Examples:
    ///     noetl query "SELECT * FROM noetl.keychain LIMIT 5"
    ///     noetl query "SELECT * FROM noetl.keychain WHERE execution_id = 123" --schema noetl
    ///     noetl query "SELECT * FROM my_table" --format json
    ///     noetl query "SELECT * FROM users" --schema public --format table
    #[command(verbatim_doc_comment)]
    Query {
        /// SQL query to execute
        query: String,

        /// Database schema (default: noetl)
        #[arg(short, long, default_value = "noetl")]
        schema: String,

        /// Output format: table or json
        #[arg(short, long, default_value = "table")]
        format: String,
    },
    /// Server management
    /// Examples:
    ///     noetl server start
    ///     noetl server start --init-db
    ///     noetl server stop
    ///     noetl server stop --force
    #[command(verbatim_doc_comment)]
    Server {
        #[command(subcommand)]
        command: ServerCommand,
    },
    /// Worker management
    /// Examples:
    ///     noetl worker start
    ///     noetl worker start --max-workers 4
    ///     noetl worker stop
    ///     noetl worker stop --name my-worker --force
    #[command(verbatim_doc_comment)]
    Worker {
        #[command(subcommand)]
        command: WorkerCommand,
    },
    /// Database management
    /// Examples:
    ///     noetl db init
    ///     noetl db validate
    #[command(verbatim_doc_comment)]
    Db {
        #[command(subcommand)]
        command: DbCommand,
    },
    /// Build Docker images
    /// Examples:
    ///     noetl build
    ///     noetl build --no-cache
    ///     noetl build --platform linux/arm64
    #[command(verbatim_doc_comment)]
    Build {
        /// Build without cache
        #[arg(long)]
        no_cache: bool,

        /// Target platform for the Docker image (e.g., linux/amd64, linux/arm64)
        #[arg(long)]
        platform: Option<String>,
    },
    /// Kubernetes deployment management
    /// Examples:
    ///     noetl k8s deploy
    ///     noetl k8s redeploy
    ///     noetl k8s reset
    ///     noetl k8s remove
    #[command(verbatim_doc_comment)]
    K8s {
        #[command(subcommand)]
        command: K8sCommand,
    },
    /// Clean up stuck executions
    /// Examples:
    ///     noetl cleanup
    ///     noetl cleanup --older-than-minutes 10
    ///     noetl cleanup --dry-run
    ///     noetl cleanup --older-than-minutes 5 --dry-run --json
    #[command(verbatim_doc_comment)]
    Cleanup {
        /// Minimum age in minutes for executions to be cancelled (default: 5)
        #[arg(long, default_value = "5")]
        older_than_minutes: u32,

        /// Dry-run mode: show what would be cancelled without actually cancelling
        #[arg(long)]
        dry_run: bool,

        /// Emit only the JSON response
        #[arg(short, long)]
        json: bool,
    },
    /// Infrastructure as Playbook (IaP) - manage cloud infrastructure using playbooks
    ///
    /// IaP provides Terraform-like infrastructure management using NoETL playbooks.
    /// State is stored locally in DuckDB and can be synced to GCS for team collaboration.
    ///
    /// Examples:
    ///     noetl iap init --project my-gcp-project --bucket my-state-bucket
    ///     noetl iap plan infrastructure.yaml
    ///     noetl iap apply infrastructure.yaml
    ///     noetl iap state list
    ///     noetl iap sync push
    ///     noetl iap drift detect
    #[command(verbatim_doc_comment)]
    Iap {
        #[command(subcommand)]
        command: IapCommand,
    },
}

#[derive(Subcommand)]
enum CatalogCommand {
    /// Register a resource (auto-detects type)
    /// Example for playbooks:
    ///     noetl catalog register tests/fixtures/playbooks/hello_world/hello_world.yaml
    ///     noetl --host=localhost --port=8082 catalog register tests/fixtures/playbooks/hello_world/hello_world.yaml
    /// Example for credential:
    ///     noetl --host=localhost --port=8082 catalog register tests/fixtures/credentials/google_oauth.json
    #[command(verbatim_doc_comment)]
    Register {
        /// Path to the resource file
        file: PathBuf,
    },
    /// Get resource details from catalog
    /// Examples:
    ///     noetl catalog get my-playbook
    ///     noetl --host=localhost --port=8082 catalog get workflows/data-pipeline
    ///     noetl catalog get my-credential
    #[command(verbatim_doc_comment)]
    Get {
        /// Resource path/name
        path: String,
    },
    /// List resources in catalog by type
    /// Examples:
    ///     noetl catalog list Playbook
    ///     noetl catalog list Credential
    ///     noetl --host=localhost --port=8082 catalog list Playbook --json
    #[command(verbatim_doc_comment)]
    List {
        /// Resource type (e.g., Playbook, Credential)
        resource_type: String,

        /// Emit only the JSON response
        #[arg(short, long)]
        json: bool,
    },
}

#[derive(Subcommand)]
enum ExecuteCommand {
    /// Execute a playbook with optional input parameters
    /// Examples:
    ///     noetl execute playbook my-playbook
    ///     noetl execute playbook workflows/etl-pipeline --input params.json
    ///     noetl --host=localhost --port=8082 execute playbook data-sync --input /path/to/input.json --json
    #[command(verbatim_doc_comment)]
    Playbook {
        /// Playbook path/name as registered in catalog
        path: String,

        /// Path to JSON file with parameters
        #[arg(short, long)]
        input: Option<PathBuf>,

        /// Emit only the JSON response
        #[arg(short, long)]
        json: bool,
    },
    /// Rerun a previous execution using the same playbook path/workload baseline.
    /// Examples:
    ///     noetl execute rerun 12345
    ///     noetl execute rerun 12345 --input params.json --json
    #[command(verbatim_doc_comment)]
    Rerun {
        /// Source execution ID to rerun
        execution_id: String,

        /// Optional JSON workload override file
        #[arg(short, long)]
        input: Option<PathBuf>,

        /// Emit only the JSON response
        #[arg(short, long)]
        json: bool,
    },
    /// Get execution status for a playbook run
    /// Examples:
    ///     noetl execute status 12345
    ///     noetl --host=localhost --port=8082 execute status 12345 --json
    #[command(verbatim_doc_comment)]
    Status {
        /// Execution ID
        execution_id: String,

        /// Emit only the JSON response
        #[arg(short, long)]
        json: bool,
    },
}

#[derive(Subcommand)]
enum GetResource {
    /// Get credential details with optional data inclusion
    /// Examples:
    ///     noetl get credential my-db-creds
    ///     noetl get credential google_oauth --include_data=false
    ///     noetl --host=localhost --port=8082 get credential aws-credentials
    #[command(verbatim_doc_comment)]
    Credential {
        /// Name of the credential
        name: String,

        /// Include decrypted data
        #[arg(long, default_value_t = true)]
        include_data: bool,
    },
}

#[derive(Subcommand)]
enum RegisterResource {
    /// Register credential(s) from JSON file or directory
    /// Examples:
    ///     noetl register credential --file credentials/postgres.json
    ///     noetl register credential --directory tests/fixtures/credentials
    ///     noetl --host=localhost --port=8082 register credential -f tests/fixtures/credentials/google_oauth.json
    #[command(verbatim_doc_comment)]
    Credential {
        /// Path to credential file
        #[arg(short, long, conflicts_with = "directory")]
        file: Option<PathBuf>,

        /// Path to directory containing credential JSON files (scans recursively)
        #[arg(short, long, conflicts_with = "file")]
        directory: Option<PathBuf>,
    },
    /// Register playbook(s) from YAML file or directory
    /// Examples:
    ///     noetl register playbook --file playbooks/my-workflow.yaml
    ///     noetl register playbook --directory tests/fixtures/playbooks
    ///     noetl --host=localhost --port=8082 register playbook -f tests/fixtures/playbooks/hello_world/hello_world.yaml
    #[command(verbatim_doc_comment)]
    Playbook {
        /// Path to playbook file
        #[arg(short, long, conflicts_with = "directory")]
        file: Option<PathBuf>,

        /// Path to directory containing playbook YAML files (scans recursively)
        #[arg(short, long, conflicts_with = "file")]
        directory: Option<PathBuf>,
    },
}

#[derive(Subcommand)]
enum ServerCommand {
    /// Start NoETL server
    /// Examples:
    ///     noetl server start
    ///     noetl server start --init-db
    #[command(verbatim_doc_comment)]
    Start {
        /// Initialize database schema on startup
        #[arg(long)]
        init_db: bool,
    },
    /// Stop NoETL server
    /// Examples:
    ///     noetl server stop
    ///     noetl server stop --force
    #[command(verbatim_doc_comment)]
    Stop {
        /// Force stop without confirmation
        #[arg(short, long)]
        force: bool,
    },
}

#[derive(Subcommand)]
enum WorkerCommand {
    /// Start NoETL worker pool
    /// Examples:
    ///     noetl worker start
    ///     noetl worker start --max-workers 4
    #[command(verbatim_doc_comment)]
    Start {
        /// Maximum number of worker threads
        #[arg(short = 'm', long)]
        max_workers: Option<usize>,
    },
    /// Stop NoETL worker
    /// Examples:
    ///     noetl worker stop
    ///     noetl worker stop --name my-worker
    ///     noetl worker stop --name my-worker --force
    #[command(verbatim_doc_comment)]
    Stop {
        /// Worker name to stop (if not specified, lists all workers)
        #[arg(short = 'n', long)]
        name: Option<String>,

        /// Force stop without confirmation
        #[arg(short, long)]
        force: bool,
    },
}

#[derive(Subcommand)]
enum DbCommand {
    /// Initialize NoETL database schema
    /// Example:
    ///     noetl db init
    #[command(verbatim_doc_comment)]
    Init,
    /// Validate NoETL database schema
    /// Example:
    ///     noetl db validate
    #[command(verbatim_doc_comment)]
    Validate,
}

#[derive(Subcommand)]
enum K8sCommand {
    /// Deploy NoETL to Kubernetes (kind cluster)
    /// Example:
    ///     noetl k8s deploy
    #[command(verbatim_doc_comment)]
    Deploy,
    /// Remove NoETL from Kubernetes
    /// Example:
    ///     noetl k8s remove
    #[command(verbatim_doc_comment)]
    Remove,
    /// Rebuild and redeploy NoETL to Kubernetes
    /// Example:
    ///     noetl k8s redeploy
    ///     noetl k8s redeploy --no-cache
    ///     noetl k8s redeploy --platform linux/arm64
    #[command(verbatim_doc_comment)]
    Redeploy {
        /// Build without cache
        #[arg(long)]
        no_cache: bool,

        /// Target platform for the Docker image (e.g., linux/amd64, linux/arm64)
        #[arg(long)]
        platform: Option<String>,
    },
    /// Reset NoETL: rebuild, redeploy, reset schema, and setup test environment
    /// Example:
    ///     noetl k8s reset
    ///     noetl k8s reset --no-cache
    ///     noetl k8s reset --platform linux/arm64
    #[command(verbatim_doc_comment)]
    Reset {
        /// Build without cache
        #[arg(long)]
        no_cache: bool,

        /// Target platform for the Docker image (e.g., linux/amd64, linux/arm64)
        #[arg(long)]
        platform: Option<String>,
    },
}

#[derive(Subcommand)]
enum IapCommand {
    /// Initialize IaP state for a project
    ///
    /// Creates local DuckDB state database and optionally configures GCS sync.
    ///
    /// Examples:
    ///     noetl iap init --project my-gcp-project
    ///     noetl iap init --project my-gcp-project --bucket my-state-bucket
    ///     noetl iap init --project my-gcp-project --region us-central1
    #[command(verbatim_doc_comment)]
    Init {
        /// GCP project ID
        #[arg(long)]
        project: String,

        /// GCS bucket for remote state (optional)
        #[arg(long)]
        bucket: Option<String>,

        /// GCP region (default: us-central1)
        #[arg(long, default_value = "us-central1")]
        region: String,

        /// Local state database path (default: .noetl/state.duckdb)
        #[arg(long, default_value = ".noetl/state.duckdb")]
        state_db: String,

        /// Workspace name for state isolation (default: default)
        #[arg(long, default_value = "default")]
        workspace: String,

        /// Remote state path template (use {workspace} placeholder)
        /// Example: workspaces/{workspace}/state.duckdb
        #[arg(long, default_value = "workspaces/{workspace}/state.duckdb")]
        state_path: String,
    },
    /// Plan infrastructure changes (dry-run)
    ///
    /// Executes the playbook in plan mode, showing what changes would be made.
    ///
    /// Examples:
    ///     noetl iap plan infrastructure.yaml
    ///     noetl iap plan gke_autopilot.yaml --var cluster_name=my-cluster
    #[command(verbatim_doc_comment)]
    Plan {
        /// Path to infrastructure playbook
        playbook: PathBuf,

        /// Set variables (format: key=value)
        #[arg(long = "var", value_name = "KEY=VALUE")]
        variables: Vec<String>,

        /// Enable verbose output
        #[arg(short, long)]
        verbose: bool,
    },
    /// Apply infrastructure changes
    ///
    /// Executes the playbook and records state changes.
    ///
    /// Examples:
    ///     noetl iap apply infrastructure.yaml
    ///     noetl iap apply gke_autopilot.yaml --var cluster_name=my-cluster
    ///     noetl iap apply infrastructure.yaml --auto-approve
    #[command(verbatim_doc_comment)]
    Apply {
        /// Path to infrastructure playbook
        playbook: PathBuf,

        /// Set variables (format: key=value)
        #[arg(long = "var", value_name = "KEY=VALUE")]
        variables: Vec<String>,

        /// Skip confirmation prompt
        #[arg(long)]
        auto_approve: bool,

        /// Enable verbose output
        #[arg(short, long)]
        verbose: bool,
    },
    /// Manage infrastructure state
    ///
    /// View, inspect, or modify the local state database.
    #[command(verbatim_doc_comment)]
    State {
        #[command(subcommand)]
        command: IapStateCommand,
    },
    /// Sync state with remote storage (GCS)
    ///
    /// Push or pull state to/from GCS for team collaboration.
    #[command(verbatim_doc_comment)]
    Sync {
        #[command(subcommand)]
        command: IapSyncCommand,
    },
    /// Detect configuration drift
    ///
    /// Compare current infrastructure state with actual cloud resources.
    ///
    /// Examples:
    ///     noetl iap drift detect
    ///     noetl iap drift detect --resource gke-cluster/my-cluster
    #[command(verbatim_doc_comment)]
    Drift {
        #[command(subcommand)]
        command: IapDriftCommand,
    },
    /// Manage workspaces for multi-developer collaboration
    ///
    /// Workspaces isolate state for different environments or developers.
    ///
    /// Examples:
    ///     noetl iap workspace list
    ///     noetl iap workspace switch staging
    ///     noetl iap workspace create dev-alice
    #[command(verbatim_doc_comment)]
    Workspace {
        #[command(subcommand)]
        command: IapWorkspaceCommand,
    },
}

#[derive(Subcommand)]
enum IapStateCommand {
    /// List all resources in state
    /// Example:
    ///     noetl iap state list
    #[command(verbatim_doc_comment)]
    List {
        /// Filter by resource type (e.g., gke-cluster, gcs-bucket)
        #[arg(long)]
        resource_type: Option<String>,

        /// Output format: table or json
        #[arg(short, long, default_value = "table")]
        format: String,
    },
    /// Show details for a specific resource
    /// Example:
    ///     noetl iap state show gke-cluster/my-cluster
    #[command(verbatim_doc_comment)]
    Show {
        /// Resource identifier (type/name)
        resource: String,
    },
    /// Remove a resource from state (does not destroy the actual resource)
    /// Example:
    ///     noetl iap state rm gke-cluster/my-cluster
    #[command(verbatim_doc_comment)]
    Rm {
        /// Resource identifier (type/name)
        resource: String,

        /// Skip confirmation prompt
        #[arg(long)]
        force: bool,
    },
    /// Execute raw SQL query against state database
    /// Example:
    ///     noetl iap state query "SELECT * FROM resources WHERE status = 'active'"
    #[command(verbatim_doc_comment)]
    Query {
        /// SQL query to execute
        sql: String,
    },
}

#[derive(Subcommand)]
enum IapSyncCommand {
    /// Push local state to GCS
    /// Example:
    ///     noetl iap sync push
    #[command(verbatim_doc_comment)]
    Push {
        /// Skip confirmation prompt
        #[arg(long)]
        force: bool,
    },
    /// Pull state from GCS to local
    /// Example:
    ///     noetl iap sync pull
    #[command(verbatim_doc_comment)]
    Pull {
        /// Skip confirmation prompt
        #[arg(long)]
        force: bool,
    },
    /// Show sync status (local vs remote)
    /// Example:
    ///     noetl iap sync status
    #[command(verbatim_doc_comment)]
    Status,
}

#[derive(Subcommand)]
enum IapDriftCommand {
    /// Detect drift between state and actual resources
    /// Example:
    ///     noetl iap drift detect
    #[command(verbatim_doc_comment)]
    Detect {
        /// Filter by resource type
        #[arg(long)]
        resource_type: Option<String>,

        /// Specific resource to check
        #[arg(long)]
        resource: Option<String>,
    },
    /// Show drift report
    /// Example:
    ///     noetl iap drift report
    #[command(verbatim_doc_comment)]
    Report {
        /// Output format: table or json
        #[arg(short, long, default_value = "table")]
        format: String,
    },
}

#[derive(Subcommand)]
enum IapWorkspaceCommand {
    /// List all registered workspaces
    /// Shows both local registry and remote workspaces from GCS
    /// Example:
    ///     noetl iap workspace list
    ///     noetl iap workspace list --remote
    #[command(verbatim_doc_comment)]
    List {
        /// Include remote workspaces from GCS
        #[arg(long)]
        remote: bool,
    },
    /// Switch to a different workspace
    /// Updates local state to use the specified workspace
    /// Example:
    ///     noetl iap workspace switch staging
    ///     noetl iap workspace switch dev-alice --pull
    #[command(verbatim_doc_comment)]
    Switch {
        /// Workspace name to switch to
        name: String,

        /// Pull state from remote after switching
        #[arg(long)]
        pull: bool,
    },
    /// Show current workspace info
    /// Example:
    ///     noetl iap workspace current
    #[command(verbatim_doc_comment)]
    Current,
    /// Create a new workspace
    /// Example:
    ///     noetl iap workspace create dev-bob
    ///     noetl iap workspace create dev-charlie --from staging
    #[command(verbatim_doc_comment)]
    Create {
        /// New workspace name
        name: String,

        /// Clone state from existing workspace
        #[arg(long)]
        from: Option<String>,

        /// Switch to new workspace after creation
        #[arg(long)]
        switch: bool,
    },
    /// Delete a workspace from registry (does not delete remote state)
    /// Example:
    ///     noetl iap workspace delete dev-old
    #[command(verbatim_doc_comment)]
    Delete {
        /// Workspace name to delete
        name: String,

        /// Also delete remote state from GCS
        #[arg(long)]
        remote: bool,

        /// Skip confirmation prompt
        #[arg(long)]
        force: bool,
    },
}

#[derive(Subcommand)]
enum ContextCommand {
    /// Add a new context for connecting to NoETL servers
    /// Examples:
    ///     noetl context add local --server-url=http://localhost:8082
    ///     noetl context add local-dev --server-url=http://localhost:8082 --runtime=local
    ///     noetl context add prod --server-url=https://noetl.example.com --runtime=distributed
    ///     noetl context add gateway --server-url=https://gateway.example.com --auth0-domain=tenant.auth0.com --auth0-client-id=abc123 --auth0-redirect-uri=https://app.example.com/login --auth0-audience=https://api.example.com
    ///     noetl context add staging --server-url=http://staging:8082 --set-current
    #[command(verbatim_doc_comment)]
    Add {
        /// Context name
        name: String,
        /// Server URL (e.g., http://localhost:8082)
        #[arg(long)]
        server_url: String,
        /// Default runtime mode for this context: local, distributed, or auto
        #[arg(long, default_value = "auto")]
        runtime: String,
        /// Default Auth0 domain to use for 'noetl auth login'
        #[arg(long)]
        auth0_domain: Option<String>,
        /// Auth0 application client_id (used to build the browser authorization URL)
        #[arg(long)]
        auth0_client_id: Option<String>,
        /// Auth0 redirect URI after browser login (e.g. https://app.example.com/login)
        #[arg(long)]
        auth0_redirect_uri: Option<String>,
        /// Optional Auth0 audience for password grant token exchange
        #[arg(long)]
        auth0_audience: Option<String>,
        /// Auth0 client_secret for password grant login (stored locally)
        #[arg(long)]
        auth0_client_secret: Option<String>,
        /// Kubernetes context name for ``noetl context port-forward``
        /// (e.g. ``gke_my-project_us-central1_my-cluster``)
        #[arg(long)]
        kube_context: Option<String>,
        /// Kubernetes namespace for ``noetl context port-forward``
        #[arg(long)]
        kube_namespace: Option<String>,
        /// Kubernetes service name (default ``noetl``)
        #[arg(long)]
        kube_service: Option<String>,
        /// In-cluster port for the service (default 8082)
        #[arg(long)]
        kube_remote_port: Option<u16>,
        /// Set as current context
        #[arg(long)]
        set_current: bool,
    },
    /// Bootstrap a context by reading the gateway's runtime contract.
    /// GETs ``<from-gateway>/api/runtime/contract``, parses any
    /// ``auth0`` block, prompts the operator to confirm the
    /// discovered Auth0 settings, and writes the context.  When the
    /// gateway doesn't carry an ``auth0`` block (older gateway image
    /// or non-Auth0 deployment) the context is created with
    /// ``server_url`` only and a warning is printed.
    ///
    /// Examples:
    ///     noetl context init gke-prod --from-gateway https://gateway.mestumre.dev
    ///     noetl context init gke-prod --from-gateway https://gateway.mestumre.dev --yes
    ///     noetl context init gke-prod --from-gateway https://gateway.mestumre.dev --set-current
    #[command(verbatim_doc_comment)]
    Init {
        /// Context name to create (replaces any existing entry with the same name)
        name: String,
        /// Gateway URL to discover Auth0 + server config from
        #[arg(long)]
        from_gateway: String,
        /// Skip the confirmation prompt (useful for scripts / CI)
        #[arg(long = "yes", visible_alias = "non-interactive")]
        non_interactive: bool,
        /// Default runtime mode for the new context: local, distributed, or auto
        #[arg(long, default_value = "distributed")]
        runtime: String,
        /// Set as current context after writing
        #[arg(long)]
        set_current: bool,
    },
    /// Update an existing context's settings in place.  All flags are
    /// optional; unspecified fields are preserved.  Refuses if the
    /// context does not exist (suggests ``context add``).
    /// Examples:
    ///     noetl context update gke-prod --auth0-client-id=abc123
    ///     noetl context update gke-pf --server-url=http://127.0.0.1:18082
    ///     noetl context update local --runtime=local
    ///     noetl context update gke-prod --auth0-audience=""   # clear
    #[command(verbatim_doc_comment)]
    Update {
        /// Context name (must already exist)
        name: String,
        /// New server URL
        #[arg(long)]
        server_url: Option<String>,
        /// New default runtime mode: local, distributed, or auto
        #[arg(long)]
        runtime: Option<String>,
        /// New Auth0 domain (empty string clears)
        #[arg(long)]
        auth0_domain: Option<String>,
        /// New Auth0 client_id (empty string clears)
        #[arg(long)]
        auth0_client_id: Option<String>,
        /// New Auth0 redirect URI (empty string clears)
        #[arg(long)]
        auth0_redirect_uri: Option<String>,
        /// New Auth0 audience (empty string clears)
        #[arg(long)]
        auth0_audience: Option<String>,
        /// New Auth0 client_secret (empty string clears)
        #[arg(long)]
        auth0_client_secret: Option<String>,
        /// New Kubernetes context name for ``noetl context port-forward``
        /// (empty string clears).
        #[arg(long)]
        kube_context: Option<String>,
        /// New Kubernetes namespace (empty string clears).
        #[arg(long)]
        kube_namespace: Option<String>,
        /// New Kubernetes service name (empty string clears, defaults to ``noetl``).
        #[arg(long)]
        kube_service: Option<String>,
        /// New in-cluster port (0 clears, defaults to 8082).
        #[arg(long)]
        kube_remote_port: Option<u16>,
    },
    /// List all configured contexts
    /// Example:
    ///     noetl context list
    #[command(verbatim_doc_comment)]
    List,
    /// Switch to a different context
    /// Examples:
    ///     noetl context use local
    ///     noetl context use prod
    #[command(verbatim_doc_comment)]
    Use {
        /// Context name to switch to
        name: String,
    },
    /// Set runtime mode for current context
    /// Examples:
    ///     noetl context set-runtime local
    ///     noetl context set-runtime distributed
    ///     noetl context set-runtime auto
    #[command(verbatim_doc_comment)]
    SetRuntime {
        /// Runtime mode: local, distributed, or auto
        runtime: String,
    },
    /// Delete a context
    /// Examples:
    ///     noetl context delete old-env
    ///     noetl context delete staging
    #[command(verbatim_doc_comment)]
    Delete {
        /// Context name to delete
        name: String,
    },
    /// Show current active context
    /// Example:
    ///     noetl context current
    #[command(verbatim_doc_comment)]
    Current,
    /// Start, stop, or check a ``kubectl port-forward`` managed by noetl
    /// for the named context.  Requires the context to have
    /// ``kube_context`` + ``kube_namespace`` set first via
    /// ``noetl context update <name> --kube-context=... --kube-namespace=...``.
    /// Examples:
    ///     noetl context port-forward gke-pf             # foreground, exits on Ctrl-C
    ///     noetl context port-forward gke-pf --detach    # background daemon, writes PID file
    ///     noetl context port-forward gke-pf --stop      # kill the daemon
    ///     noetl context port-forward gke-pf --status    # show daemon PID + liveness
    #[command(verbatim_doc_comment)]
    PortForward {
        /// Context name (must already exist with ``kube_context`` + ``kube_namespace`` set)
        name: String,
        /// Spawn the port-forward as a background daemon.  Writes the PID
        /// to ``~/.config/noetl/port-forwards/<context>.pid`` so a later
        /// ``--stop`` can find it.
        #[arg(long, conflicts_with_all = ["stop", "status"])]
        detach: bool,
        /// Stop a previously-detached port-forward by reading the PID file
        /// and sending SIGTERM, then SIGKILL after a short grace.
        #[arg(long, conflicts_with_all = ["detach", "status"])]
        stop: bool,
        /// Report whether a detached port-forward exists + its PID.
        #[arg(long, conflicts_with_all = ["detach", "stop"])]
        status: bool,
    },
}

#[derive(Subcommand)]
enum AuthCommand {
    /// Login via Gateway /api/auth/login and optionally cache session token in context
    /// Examples:
    ///     noetl auth login --auth0-token <ID_TOKEN>
    ///     noetl auth login --auth0-callback-url 'https://app/login#id_token=...'
    ///     noetl auth login --auth0 user@example.com
    ///     noetl auth login --auth0 user@example.com --password
    ///     noetl auth login --auth0 user@example.com --password --auth0-audience https://api.example.com
    ///     noetl auth login --auth0 user@example.com --password --auth0-client-secret $AUTH0_CLIENT_SECRET
    ///     noetl auth login --browser
    ///     noetl auth login --browser-pkce
    ///     noetl auth login --browser-pkce --pkce-port 8765 --auth0 user@example.com
    ///     noetl auth login --auth0-token <ID_TOKEN> --context gateway
    #[command(verbatim_doc_comment)]
    Login {
        /// Unified Auth0 input: email/login hint, raw token, or callback URL
        /// If email is provided, CLI prompts for password (password grant) or callback URL/token.
        #[arg(long, value_name = "EMAIL|TOKEN|CALLBACK_URL", conflicts_with_all = ["auth0_token", "auth0_callback_url"])]
        auth0: Option<String>,

        /// Auth0 ID/access token (the value sent as auth0_token to gateway)
        #[arg(long)]
        auth0_token: Option<String>,

        /// Full Auth0 callback URL containing #id_token=... or #access_token=...
        #[arg(long)]
        auth0_callback_url: Option<String>,

        /// Auth0 domain (defaults to context value, then gateway default)
        #[arg(long)]
        auth0_domain: Option<String>,

        /// Auth0 redirect URI for browser-pkce login (must be localhost/127.0.0.1 callback)
        #[arg(long)]
        auth0_redirect_uri: Option<String>,

        /// Optional Auth0 audience for password grant (overrides context value)
        #[arg(long)]
        auth0_audience: Option<String>,

        /// Optional Auth0 client secret for password grant (overrides context and NOETL_AUTH0_CLIENT_SECRET)
        #[arg(long)]
        auth0_client_secret: Option<String>,

        /// Browser/device login flow (gcloud-style) without callback copy/paste
        #[arg(long, conflicts_with_all = ["password", "browser_pkce"])]
        browser: bool,

        /// Browser PKCE login flow with localhost callback endpoint (gcloud login style)
        #[arg(long, alias = "pkce", conflicts_with_all = ["password", "browser"])]
        browser_pkce: bool,

        /// Local callback port for --browser-pkce (default: 8765)
        #[arg(long, default_value_t = 8765)]
        pkce_port: u16,

        /// Do not auto-open browser in --browser/--browser-pkce mode (print URL only)
        #[arg(long)]
        no_browser_open: bool,

        /// Use Auth0 password grant (prompts for password in terminal, no browser needed)
        #[arg(long)]
        password: bool,

        /// Session duration in hours
        #[arg(long, default_value_t = 8)]
        session_duration_hours: i32,

        /// Context name to store session token into (defaults to current context)
        #[arg(long)]
        context: Option<String>,

        /// Do not save the returned session token into context
        #[arg(long)]
        no_save: bool,

        /// Emit only JSON response
        #[arg(short, long)]
        json: bool,
    },
    /// Remove cached gateway session token from context
    /// Example:
    ///     noetl auth logout
    ///     noetl auth logout --context gateway
    #[command(verbatim_doc_comment)]
    Logout {
        /// Context name to clear token from (defaults to current context)
        #[arg(long)]
        context: Option<String>,
    },
}

#[derive(Serialize)]
struct RegisterRequest {
    content: String,
    resource_type: String,
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
enum ExecutionResourceKind {
    Playbook,
    Notebook,
}

#[derive(Serialize)]
struct ExecuteApiRequest {
    path: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    version: Option<i64>,
    workload: serde_json::Value,
    resource_kind: ExecutionResourceKind,
}

#[derive(Serialize)]
struct RerunApiRequest {
    workload: serde_json::Value,
    resource_kind: ExecutionResourceKind,
}

/// Reference type for playbook execution
#[derive(Debug)]
enum RefType {
    /// Local file path: ./playbooks/foo.yaml
    File(PathBuf),
    /// Catalog reference: catalog://name@version
    Catalog { name: String, version: Option<String> },
    /// Database ID: pbk_01J...
    DatabaseId(String),
    /// Catalog path (requires distributed): workflows/etl-pipeline
    CatalogPath(String),
}

/// Execution context parsed from reference
#[derive(Debug)]
struct ExecContext {
    ref_type: RefType,
    version: Option<String>,
    /// Target inferred when reference is treated as target for noetl.yaml
    inferred_target: Option<String>,
}

/// Parse execution reference into type and metadata
fn parse_exec_reference(reference: &str, version_override: Option<&str>) -> Result<ExecContext> {
    // Pattern 1: catalog://name@version
    if reference.starts_with("catalog://") {
        let rest = reference.strip_prefix("catalog://").unwrap();
        let (name, version) = if let Some(at_pos) = rest.find('@') {
            (rest[..at_pos].to_string(), Some(rest[at_pos + 1..].to_string()))
        } else {
            (rest.to_string(), version_override.map(|s| s.to_string()))
        };
        return Ok(ExecContext {
            ref_type: RefType::Catalog {
                name,
                version: version.clone(),
            },
            version,
            inferred_target: None,
        });
    }

    // Pattern 2: Database ID (pbk_xxx or similar UUID-like)
    if reference.starts_with("pbk_")
        || (reference.len() > 20 && reference.chars().all(|c| c.is_alphanumeric() || c == '_' || c == '-'))
    {
        return Ok(ExecContext {
            ref_type: RefType::DatabaseId(reference.to_string()),
            version: version_override.map(|s| s.to_string()),
            inferred_target: None,
        });
    }

    // Pattern 3: File path (contains / or \ or ends with .yaml/.yml)
    // IMPORTANT: Only treat as file when it's an actual file. A directory path
    // (e.g. catalog-like path with '/') should continue to catalog resolution.
    let is_file_like = reference.contains('/')
        || reference.contains('\\')
        || reference.ends_with(".yaml")
        || reference.ends_with(".yml");

    if is_file_like {
        let path = PathBuf::from(reference);
        if path.is_file() || reference.ends_with(".yaml") || reference.ends_with(".yml") {
            return Ok(ExecContext {
                ref_type: RefType::File(path),
                version: None,
                inferred_target: None,
            });
        }
    }

    // Try to find as file with extension
    let try_find_file = |base: &str| -> Option<PathBuf> {
        let path = PathBuf::from(base);
        if path.exists() && path.is_file() {
            return Some(path);
        }
        let yaml_path = PathBuf::from(format!("{}.yaml", base));
        if yaml_path.exists() && yaml_path.is_file() {
            return Some(yaml_path);
        }
        let yml_path = PathBuf::from(format!("{}.yml", base));
        if yml_path.exists() && yml_path.is_file() {
            return Some(yml_path);
        }
        // Try automation/ prefix as fallback
        let auto_yaml = PathBuf::from(format!("automation/{}.yaml", base));
        if auto_yaml.exists() && auto_yaml.is_file() {
            return Some(auto_yaml);
        }
        let auto_yml = PathBuf::from(format!("automation/{}.yml", base));
        if auto_yml.exists() && auto_yml.is_file() {
            return Some(auto_yml);
        }
        None
    };

    if let Some(file_path) = try_find_file(reference) {
        return Ok(ExecContext {
            ref_type: RefType::File(file_path),
            version: None,
            inferred_target: None,
        });
    }

    // Pattern 5: Check for noetl.yaml in current directory - treat reference as target
    // Only apply for simple target names. If reference contains path separators,
    // prefer catalog-path resolution (Pattern 6).
    let noetl_yaml = PathBuf::from("noetl.yaml");
    let is_simple_target = !reference.contains('/')
        && !reference.contains('\\')
        && !reference.ends_with(".yaml")
        && !reference.ends_with(".yml");
    if is_simple_target && noetl_yaml.exists() && noetl_yaml.is_file() {
        return Ok(ExecContext {
            ref_type: RefType::File(noetl_yaml),
            version: None,
            inferred_target: Some(reference.to_string()),
        });
    }

    // Pattern 6: Catalog path (no file found, assume it's a catalog reference)
    Ok(ExecContext {
        ref_type: RefType::CatalogPath(reference.to_string()),
        version: version_override.map(|s| s.to_string()),
        inferred_target: None,
    })
}

/// Resolve runtime mode based on:
/// 1. CLI flag (--runtime local|distributed) - highest priority if explicitly set
/// 2. If --runtime auto (default): use context config runtime preference
/// 3. If context runtime is also auto: auto-detect from reference type
fn resolve_runtime(
    runtime_flag: &str,
    context_runtime: Option<&str>,
    ctx: &ExecContext,
    verbose: bool,
) -> Result<String> {
    // CLI flag takes precedence if explicitly set (not "auto")
    if runtime_flag != "auto" {
        if verbose {
            println!("Runtime: {} (from --runtime flag)", runtime_flag);
        }
        return Ok(runtime_flag.to_string());
    }

    // --runtime auto (default): check context config preference
    if let Some(ctx_runtime) = context_runtime {
        if ctx_runtime != "auto" {
            if verbose {
                println!("Runtime: {} (from context config)", ctx_runtime);
            }
            return Ok(ctx_runtime.to_string());
        }
    }

    // Context runtime is also "auto": auto-resolve based on reference type
    let resolved = match &ctx.ref_type {
        RefType::File(_) => "local",
        RefType::Catalog { .. } => "distributed",
        RefType::DatabaseId(_) => "distributed",
        RefType::CatalogPath(_) => "distributed",
    };
    if verbose {
        println!("Runtime: {} (auto-detected from reference type)", resolved);
    }
    Ok(resolved.to_string())
}

/// Build variables map from payload JSON and --set flags
fn build_variables(payload: Option<&str>, variables: &[String]) -> Result<HashMap<String, String>> {
    let mut vars = HashMap::new();

    // Parse payload JSON if provided
    if let Some(payload_str) = payload {
        match serde_json::from_str::<serde_json::Value>(payload_str) {
            Ok(serde_json::Value::Object(map)) => {
                for (key, value) in map {
                    let value_str = match value {
                        serde_json::Value::String(s) => s,
                        serde_json::Value::Number(n) => n.to_string(),
                        serde_json::Value::Bool(b) => b.to_string(),
                        serde_json::Value::Null => "null".to_string(),
                        other => serde_json::to_string(&other).unwrap_or_else(|_| "null".to_string()),
                    };
                    vars.insert(key, value_str);
                }
            }
            Ok(_) => {
                eprintln!("Error: Payload must be a JSON object, not array or primitive");
                std::process::exit(1);
            }
            Err(e) => {
                eprintln!("Error: Invalid JSON payload: {}", e);
                std::process::exit(1);
            }
        }
    }

    // Parse --set variables (override payload)
    for var in variables {
        let parts: Vec<&str> = var.splitn(2, '=').collect();
        if parts.len() == 2 {
            vars.insert(parts[0].to_string(), parts[1].to_string());
        } else {
            eprintln!("Warning: Invalid variable format '{}', expected key=value", var);
        }
    }

    Ok(vars)
}

/// Build distributed execution payload from input file, --payload JSON, and --set overrides.
/// Merge order (last wins): input file -> --payload -> --set.
fn build_distributed_payload(
    input: Option<&PathBuf>,
    payload: Option<&str>,
    variables: &[String],
) -> Result<serde_json::Value> {
    let mut merged = serde_json::Map::new();

    if let Some(input_file) = input {
        let content = fs::read_to_string(input_file).context(format!("Failed to read input file: {:?}", input_file))?;
        let input_json: serde_json::Value = serde_json::from_str(&content).context("Failed to parse input JSON")?;
        match input_json {
            serde_json::Value::Object(map) => {
                for (k, v) in map {
                    merged.insert(k, v);
                }
            }
            _ => bail!("Input file JSON must be an object"),
        }
    }

    if let Some(payload_str) = payload {
        let payload_json: serde_json::Value =
            serde_json::from_str(payload_str).context("Failed to parse --payload JSON")?;
        match payload_json {
            serde_json::Value::Object(map) => {
                for (k, v) in map {
                    merged.insert(k, v);
                }
            }
            _ => bail!("--payload must be a JSON object"),
        }
    }

    for var in variables {
        let parts: Vec<&str> = var.splitn(2, '=').collect();
        if parts.len() != 2 {
            eprintln!("Warning: Invalid variable format '{}', expected key=value", var);
            continue;
        }

        let key = parts[0].strip_prefix("workload.").unwrap_or(parts[0]);
        let raw = parts[1];
        let value = serde_json::from_str::<serde_json::Value>(raw)
            .unwrap_or_else(|_| serde_json::Value::String(raw.to_string()));
        insert_json_path(&mut merged, key, value);
    }

    Ok(serde_json::Value::Object(merged))
}

fn insert_json_path(map: &mut serde_json::Map<String, serde_json::Value>, key: &str, value: serde_json::Value) {
    let mut parts = key.split('.').filter(|part| !part.is_empty()).peekable();
    let Some(first) = parts.next() else {
        return;
    };

    if parts.peek().is_none() {
        map.insert(first.to_string(), value);
        return;
    }

    let mut current = map
        .entry(first.to_string())
        .or_insert_with(|| serde_json::Value::Object(serde_json::Map::new()));

    while let Some(part) = parts.next() {
        let is_leaf = parts.peek().is_none();
        if !current.is_object() {
            *current = serde_json::Value::Object(serde_json::Map::new());
        }

        let object = current.as_object_mut().expect("current value forced to object");
        if is_leaf {
            object.insert(part.to_string(), value);
            return;
        }

        current = object
            .entry(part.to_string())
            .or_insert_with(|| serde_json::Value::Object(serde_json::Map::new()));
    }
}

/// Resolve playbook file path and optional target step
/// Implements File-First Strategy:
/// 1. Check if first arg is explicit path (contains / or \ or ends with .yaml/.yml)
/// 2. Check if file exists (as-is, with .yaml, with .yml)
/// 3. Auto-discover: ./noetl.yaml (priority) → ./main.yaml (fallback)
/// 4. Treat remaining args as target step
fn resolve_playbook_and_target(
    playbook_or_target: Option<String>,
    args: Vec<String>,
) -> Result<(PathBuf, Option<String>)> {
    // Helper: Check if string looks like a file path
    let is_explicit_path =
        |s: &str| -> bool { s.contains('/') || s.contains('\\') || s.ends_with(".yaml") || s.ends_with(".yml") };

    // Helper: Try to find file with extensions
    let try_find_file = |base: &str| -> Option<PathBuf> {
        // Try as-is
        let path = PathBuf::from(base);
        if path.exists() && path.is_file() {
            return Some(path);
        }

        // Try with .yaml extension
        let yaml_path = PathBuf::from(format!("{}.yaml", base));
        if yaml_path.exists() && yaml_path.is_file() {
            return Some(yaml_path);
        }

        // Try with .yml extension
        let yml_path = PathBuf::from(format!("{}.yml", base));
        if yml_path.exists() && yml_path.is_file() {
            return Some(yml_path);
        }

        None
    };

    // Helper: Auto-discover playbook in current directory
    let auto_discover = || -> Result<PathBuf> {
        // Priority 1: noetl.yaml
        if let Some(path) = try_find_file("./noetl") {
            return Ok(path);
        }

        // Priority 2: main.yaml
        if let Some(path) = try_find_file("./main") {
            return Ok(path);
        }

        // No playbook found
        anyhow::bail!(
            "No playbook found. Expected ./noetl.yaml or ./main.yaml in current directory.\n\
             Hint: Specify explicit path like: noetl run automation/main.yaml"
        )
    };

    // Case 1: No first arg provided → auto-discover, first trailing arg is target
    if playbook_or_target.is_none() {
        let playbook = auto_discover()?;
        let target = args.first().map(|s| s.to_string());
        return Ok((playbook, target));
    }

    let first_arg = playbook_or_target.unwrap();

    // Case 2: First arg is explicit path → use it, first trailing arg is target
    if is_explicit_path(&first_arg) {
        let playbook = PathBuf::from(&first_arg);
        if !playbook.exists() {
            anyhow::bail!("Playbook file not found: {}", playbook.display());
        }
        let target = args.first().map(|s| s.to_string());
        return Ok((playbook, target));
    }

    // Case 3: First arg might be a file (without extension) → check if exists
    if let Some(playbook) = try_find_file(&first_arg) {
        // File found! First trailing arg is target
        let target = args.first().map(|s| s.to_string());
        return Ok((playbook, target));
    }

    // Case 4: First arg is not a file → auto-discover, first arg is target
    let playbook = auto_discover()?;
    let target = Some(first_arg);
    Ok((playbook, target))
}

fn env_flag(name: &str) -> bool {
    std::env::var(name)
        .map(|value| {
            let normalized = value.trim().to_ascii_lowercase();
            matches!(normalized.as_str(), "1" | "true" | "yes" | "on")
        })
        .unwrap_or(false)
}

fn is_gateway_url(base_url: &str) -> bool {
    let normalized = base_url.trim().to_ascii_lowercase();
    normalized.contains("gateway.")
}

fn api_url(base_url: &str, path: &str, use_gateway_proxy: bool) -> String {
    let base = base_url.trim_end_matches('/');
    let endpoint = path.trim_start_matches('/');
    if use_gateway_proxy {
        format!("{}/noetl/{}", base, endpoint)
    } else {
        format!("{}/api/{}", base, endpoint)
    }
}

/// Resolve which named context to use for this invocation, honouring
/// the ``--context <name>`` global flag.  When the flag is set and
/// the named context doesn't exist, fail fast with a clear error —
/// silently falling back to the current context would mask the
/// typo and run the command against the wrong cluster.
///
/// Takes the override as ``Option<&str>`` (rather than ``&Cli``) so
/// callers can use it inside ``match`` arms that have already
/// destructured ``cli``.
///
/// Returns ``Some((name, context))`` when ``--context`` is honoured,
/// or ``None`` when the caller should fall back to
/// ``Config::get_current_context``.
fn resolve_named_context<'a>(
    requested: Option<&str>,
    config: &'a Config,
) -> Option<(&'a String, &'a Context)> {
    let requested = requested?;
    match config.contexts.get_key_value(requested) {
        Some((name, ctx)) => Some((name, ctx)),
        None => {
            eprintln!(
                "Context '{}' (from --context flag) not found.",
                requested
            );
            eprintln!("  Configured contexts:  noetl context list");
            eprintln!(
                "  Create it:            noetl context add {} --server-url=...",
                requested
            );
            std::process::exit(1);
        }
    }
}

/// Effective context for this invocation — ``--context <name>`` if
/// passed (validated by ``resolve_named_context``), otherwise
/// whatever ``Config`` reports as current.
fn effective_context<'a>(
    requested: Option<&str>,
    config: &'a Config,
) -> Option<(&'a String, &'a Context)> {
    resolve_named_context(requested, config).or_else(|| config.get_current_context())
}

fn resolve_base_url(cli: &Cli, config: &Config) -> String {
    if let Some(url) = cli.server_url.clone() {
        url
    } else if let (Some(host), Some(port)) = (cli.host.as_ref(), cli.port) {
        format!("http://{}:{}", host, port)
    } else {
        effective_context(cli.context.as_deref(), config)
            .map(|(_, ctx)| ctx.server_url.clone())
            .unwrap_or_else(|| "http://localhost:8082".to_string())
    }
}

fn resolve_session_token(cli: &Cli, config: &Config) -> Option<String> {
    let env_session_token = std::env::var("NOETL_SESSION_TOKEN")
        .ok()
        .filter(|v| !v.trim().is_empty());
    let context_session_token = effective_context(cli.context.as_deref(), config)
        .and_then(|(_, ctx)| ctx.gateway_session_token.clone())
        .filter(|v| !v.trim().is_empty());

    cli.session_token
        .clone()
        .or(env_session_token)
        .or(context_session_token)
}

fn display_endpoint(base_url: &str) -> String {
    base_url
        .trim()
        .trim_end_matches('/')
        .strip_prefix("https://")
        .or_else(|| base_url.trim().trim_end_matches('/').strip_prefix("http://"))
        .unwrap_or(base_url.trim().trim_end_matches('/'))
        .to_string()
}

fn split_console_args(line: &str) -> Result<Vec<String>> {
    let mut args = Vec::new();
    let mut current = String::new();
    let mut in_single = false;
    let mut in_double = false;
    let mut escaped = false;

    for ch in line.chars() {
        if escaped {
            current.push(ch);
            escaped = false;
            continue;
        }
        match ch {
            '\\' if !in_single => {
                escaped = true;
            }
            '\'' if !in_double => {
                in_single = !in_single;
            }
            '"' if !in_single => {
                in_double = !in_double;
            }
            c if c.is_whitespace() && !in_single && !in_double => {
                if !current.is_empty() {
                    args.push(std::mem::take(&mut current));
                }
            }
            _ => current.push(ch),
        }
    }

    if escaped {
        current.push('\\');
    }
    if in_single || in_double {
        anyhow::bail!("Unterminated quote in command.");
    }
    if !current.is_empty() {
        args.push(current);
    }
    Ok(args)
}

fn inject_global_cli_overrides(mut command: Command, cli: &Cli) -> Command {
    if let Some(host) = &cli.host {
        command.arg("--host").arg(host);
    }
    if let Some(port) = cli.port {
        command.arg("--port").arg(port.to_string());
    }
    if let Some(server_url) = &cli.server_url {
        command.arg("--server-url").arg(server_url);
    }
    if cli.gateway {
        command.arg("--gateway");
    }
    if let Some(token) = &cli.session_token {
        command.arg("--session-token").arg(token);
    }
    command
}

fn print_console_help() {
    println!("NoETL Console commands:");
    println!("  <noetl subcommand>   Run any CLI command without leaving console");
    println!("  help                 Show console help");
    println!("  where                Show current context/server/proxy mode");
    println!("  exit | quit          Leave console");
    println!();
    println!("Examples:");
    println!("  context list");
    println!("  context use gke-prod");
    println!("  auth login --browser-pkce");
    println!("  --gateway catalog register tests/fixtures/playbooks/foo.yaml");
    println!("  exec tests/quantum/cudaq_ai_pipeline -r distributed");
}

fn print_console_where(cli: &Cli) -> Result<()> {
    let config = Config::load()?;
    let base_url = resolve_base_url(cli, &config);
    let use_gateway_proxy = cli.gateway || env_flag("NOETL_USE_GATEWAY") || is_gateway_url(&base_url);
    let context = config.current_context.clone().unwrap_or_else(|| "-".to_string());
    let token = resolve_session_token(cli, &config);
    println!("Context: {}", context);
    println!("Server:  {}", base_url);
    println!(
        "Proxy:   {}",
        if use_gateway_proxy {
            "gateway (/noetl/*)"
        } else {
            "direct (/api/*)"
        }
    );
    println!("Token:   {}", if token.is_some() { "available" } else { "not set" });
    Ok(())
}

fn run_console(cli: &Cli) -> Result<()> {
    println!("NoETL Console started. Type 'help' for commands, 'exit' to quit.");
    loop {
        let config = Config::load()?;
        let base_url = resolve_base_url(cli, &config);
        let use_gateway_proxy = cli.gateway || env_flag("NOETL_USE_GATEWAY") || is_gateway_url(&base_url);
        let context_name = config.current_context.unwrap_or_else(|| "-".to_string());
        let endpoint = display_endpoint(&base_url);
        let mode = if use_gateway_proxy { "gw" } else { "api" };
        let prompt = format!("noetl({}@{}|{})> ", context_name, endpoint, mode);

        let line = prompt_input(&prompt)?;
        let input = line.trim();
        if input.is_empty() {
            continue;
        }
        if matches!(input, "exit" | "quit") {
            break;
        }
        if input == "help" {
            print_console_help();
            continue;
        }
        if input == "where" {
            print_console_where(cli)?;
            continue;
        }

        let mut args = split_console_args(input)?;
        if args.is_empty() {
            continue;
        }

        if args[0] == "noetl" || args[0] == "ntl" {
            args.remove(0);
        }
        if args.is_empty() {
            continue;
        }
        if args[0] == "console" {
            eprintln!("Already in console. Use 'exit' to leave.");
            continue;
        }

        // Allow one-line explicit gateway override:
        //   --gateway <command ...>
        let mut force_gateway = false;
        if args[0] == "--gateway" {
            force_gateway = true;
            args.remove(0);
            if args.is_empty() {
                eprintln!("No command provided after --gateway.");
                continue;
            }
        }

        let exe = std::env::current_exe().context("Failed to resolve current executable path")?;
        let mut child = inject_global_cli_overrides(Command::new(exe), cli);
        if force_gateway {
            child.arg("--gateway");
        }
        child.args(&args);

        let status = child.status().context("Failed to execute command in console")?;
        if !status.success() {
            eprintln!("Command exited with status: {}", status);
        }
    }

    Ok(())
}

enum CodexDispatch {
    Doctor,
    Passthrough(Vec<String>),
}

fn resolve_codex_dispatch(args: &[String]) -> CodexDispatch {
    if args.len() == 1 && args[0] == "doctor" {
        CodexDispatch::Doctor
    } else {
        CodexDispatch::Passthrough(args.to_vec())
    }
}

fn run_codex_doctor() -> Result<()> {
    let version = Command::new("codex")
        .arg("--version")
        .output()
        .context("Failed to execute 'codex --version'. Is codex installed and on PATH?")?;

    if !version.status.success() {
        let stderr = String::from_utf8_lossy(&version.stderr);
        anyhow::bail!("codex is installed but --version failed: {}", stderr.trim());
    }

    let version_out = String::from_utf8_lossy(&version.stdout).trim().to_string();
    println!(
        "codex: {}",
        if version_out.is_empty() {
            "installed"
        } else {
            &version_out
        }
    );

    // Best-effort auth check (subcommand shape may vary by codex version).
    let auth = Command::new("codex").args(["auth", "status"]).output();
    match auth {
        Ok(output) if output.status.success() => {
            let text = String::from_utf8_lossy(&output.stdout);
            if text.trim().is_empty() {
                println!("auth: ok");
            } else {
                println!("auth:\n{}", text.trim());
            }
        }
        Ok(output) => {
            let stderr = String::from_utf8_lossy(&output.stderr);
            let note = if stderr.trim().is_empty() {
                "subcommand unavailable".to_string()
            } else {
                stderr.trim().to_string()
            };
            println!("auth: check unavailable ({})", note);
        }
        Err(_) => {
            println!("auth: check unavailable");
        }
    }

    println!("workspace: {}", std::env::current_dir()?.display());
    Ok(())
}

fn run_codex_passthrough(args: &[String]) -> Result<()> {
    let status = Command::new("codex")
        .args(args)
        .status()
        .context("Failed to execute codex. Is codex installed and on PATH?")?;

    if let Some(code) = status.code() {
        std::process::exit(code);
    }

    anyhow::bail!("codex terminated by signal")
}

fn run_ai_mode(cli: &Cli, args: &[String]) -> Result<()> {
    let config = Config::load()?;
    let base_url = resolve_base_url(cli, &config);
    let mode = if cli.gateway || env_flag("NOETL_USE_GATEWAY") || is_gateway_url(&base_url) {
        "gateway"
    } else {
        "direct-api"
    };

    eprintln!("NoETL AI mode (M1 scaffold)");
    eprintln!("  endpoint: {}", base_url);
    eprintln!("  routing:  {}", mode);
    eprintln!("  note: full NoETL tool bridge and retrieval are in progress.");

    run_codex_passthrough(args)
}

/// Exit code emitted when a gateway-proxied request fails with HTTP
/// 401 because the cached session token is no longer accepted by
/// the gateway.  Distinct from the generic ``1`` failure exit so
/// shell scripts can branch on it (``[ $? -eq 77 ] && noetl auth
/// login --browser-pkce``).
pub const GATEWAY_AUTH_EXPIRED_EXIT_CODE: i32 = 77;

/// Inspect a gateway-proxied HTTP response.  When the gateway
/// returned 401, print a structured hint pointing the operator at
/// ``noetl auth login`` and exit with
/// ``GATEWAY_AUTH_EXPIRED_EXIT_CODE`` so calling scripts can detect
/// the credential-rotation case cleanly.
///
/// Pass ``use_gateway_proxy = false`` for direct-server-URL calls
/// (e.g. ``http://localhost:8082`` via kubectl port-forward); the
/// hint is only relevant when the gateway middleware is gating the
/// request.  When called against a 200..400 response or a non-401
/// failure, returns without action so the existing error path can
/// handle it.
fn check_gateway_auth_expired(
    status: reqwest::StatusCode,
    use_gateway_proxy: bool,
    current_context_name: Option<&str>,
) {
    if !use_gateway_proxy || status != reqwest::StatusCode::UNAUTHORIZED {
        return;
    }
    eprintln!();
    eprintln!("Cached gateway session token expired (HTTP 401).");
    if let Some(name) = current_context_name {
        eprintln!("  Run: noetl auth login --browser-pkce --context {}", name);
    } else {
        eprintln!("  Run: noetl auth login --browser-pkce");
        eprintln!("  (or set a current context first: noetl context use <name>)");
    }
    eprintln!();
    eprintln!(
        "Exit code {} is reserved for this case so shell scripts can detect it.",
        GATEWAY_AUTH_EXPIRED_EXIT_CODE
    );
    std::process::exit(GATEWAY_AUTH_EXPIRED_EXIT_CODE);
}

fn build_http_client(session_token: Option<&str>) -> Result<Client> {
    if let Some(token) = session_token {
        let mut headers = HeaderMap::new();
        let bearer = format!("Bearer {}", token);
        headers.insert(
            AUTHORIZATION,
            HeaderValue::from_str(&bearer).context("Invalid session token for Authorization header")?,
        );
        headers.insert(
            "x-session-token",
            HeaderValue::from_str(token).context("Invalid session token for x-session-token header")?,
        );

        Ok(Client::builder().default_headers(headers).build()?)
    } else {
        Ok(Client::new())
    }
}

#[tokio::main]
async fn main() -> Result<()> {
    let cli = Cli::parse();
    let mut config = Config::load()?;

    if matches!(cli.command, Some(Commands::Console)) {
        return run_console(&cli);
    }

    let base_url = resolve_base_url(&cli, &config);

    let session_token = resolve_session_token(&cli, &config);

    // Capture the ``--context <name>`` override as an owned Option<String>
    // before the match below destructures fields out of ``cli``.  Without
    // this binding, sites inside the match arms that want to call
    // ``effective_context(...)`` for runtime preference can't borrow
    // ``cli`` because some fields have been partially moved.
    let context_override: Option<String> = cli.context.clone();

    let use_gateway_proxy = cli.gateway || env_flag("NOETL_USE_GATEWAY") || is_gateway_url(&base_url);
    let is_auth_login_command = matches!(
        &cli.command,
        Some(Commands::Auth {
            command: AuthCommand::Login { .. }
        })
    );
    let is_context_or_auth_command = matches!(
        &cli.command,
        Some(Commands::Context { .. }) | Some(Commands::Auth { .. })
    );

    if use_gateway_proxy && session_token.is_none() && !is_auth_login_command && !is_context_or_auth_command {
        eprintln!("Warning: gateway mode enabled but no session token provided.");
        eprintln!("  Set --session-token <token> or NOETL_SESSION_TOKEN to authenticate gateway requests.");
    }

    let client = build_http_client(session_token.as_deref())?;

    if cli.interactive {
        return run_tui(&base_url, client, use_gateway_proxy).await;
    }

    match cli.command {
        Some(Commands::Console) => {}
        Some(Commands::Codex { args }) => match resolve_codex_dispatch(&args) {
            CodexDispatch::Doctor => run_codex_doctor()?,
            CodexDispatch::Passthrough(args) => run_codex_passthrough(&args)?,
        },
        Some(Commands::Ai { ref args }) => {
            run_ai_mode(&cli, &args)?;
        }
        Some(Commands::Exec {
            reference,
            runtime,
            target,
            variables,
            payload,
            input,
            version,
            endpoint,
            verbose,
            dry_run,
            json,
        }) => {
            // Get context runtime preference (honours --context override).
            let context_runtime = effective_context(context_override.as_deref(), &config).map(|(_, ctx)| ctx.runtime.as_str());

            // Parse the reference to determine type and resolve runtime
            let exec_ctx = parse_exec_reference(&reference, version.as_deref())?;
            let effective_runtime = resolve_runtime(&runtime, context_runtime, &exec_ctx, verbose)?;
            let effective_endpoint = endpoint.unwrap_or_else(|| base_url.clone());
            let effective_gateway_proxy = use_gateway_proxy || is_gateway_url(&effective_endpoint);

            // Build variables from payload and --set flags
            let vars = build_variables(payload.as_deref(), &variables)?;

            if verbose {
                println!("Execution Context:");
                println!("  Reference: {}", reference);
                println!("  Type: {:?}", exec_ctx.ref_type);
                println!("  Runtime: {}", effective_runtime);
                if let Some(v) = &exec_ctx.version {
                    println!("  Version: {}", v);
                }
                if let Some(t) = &target {
                    println!("  Target: {}", t);
                } else if let Some(t) = &exec_ctx.inferred_target {
                    println!("  Target: {} (inferred from noetl.yaml)", t);
                }
            }

            if dry_run {
                println!("\n[DRY RUN] Would execute with:");
                println!("  Runtime: {}", effective_runtime);
                match &exec_ctx.ref_type {
                    RefType::File(path) => println!("  File: {}", path.display()),
                    RefType::Catalog { name, version } => {
                        println!("  Catalog: {}", name);
                        if let Some(v) = version {
                            println!("  Version: {}", v);
                        }
                    }
                    RefType::DatabaseId(id) => println!("  DB ID: {}", id),
                    RefType::CatalogPath(path) => println!("  Path: {}", path),
                }
                return Ok(());
            }

            match effective_runtime.as_str() {
                "local" => {
                    // Local execution using Rust interpreter
                    let playbook_path = match &exec_ctx.ref_type {
                        RefType::File(path) => path.clone(),
                        RefType::Catalog { .. } | RefType::DatabaseId(_) | RefType::CatalogPath(_) => {
                            eprintln!("Error: Local runtime requires a file path reference");
                            eprintln!("  Use: noetl exec ./path/to/playbook.yaml -r local");
                            eprintln!("  Or use: -r distributed for catalog/db references");
                            std::process::exit(1);
                        }
                    };

                    // Use CLI target if provided, otherwise use inferred target (from noetl.yaml pattern)
                    let effective_target = target.or(exec_ctx.inferred_target.clone());

                    // The `--json` flag now does two things in local
                    // runtime mode (previously a no-op): (1) suppress
                    // human-readable progress on stdout — progress
                    // already routes to stderr from PlaybookRunner — and
                    // (2) emit a structured RunOutcome envelope on
                    // stdout when the playbook completes. This is what
                    // lets the agent bridge (and any other programmatic
                    // consumer) get clean JSON to pipe into `jq`.
                    let runner = playbook_runner::PlaybookRunner::new(playbook_path)
                        .with_variables(vars)
                        .with_verbose(verbose)
                        .with_target(effective_target)
                        // When --json is set, suppress progress entirely
                        // so stdout = ONLY the JSON envelope. The user
                        // still sees errors via stderr.
                        .with_quiet(json)
                        .with_emit_json(json);

                    runner.run()?;
                }
                "distributed" => {
                    // Distributed execution via server
                    let (path, version) = match &exec_ctx.ref_type {
                        RefType::File(file_path) => {
                            // For file refs, we need to register first or use a different approach
                            eprintln!("Warning: Executing local file via distributed runtime");
                            eprintln!(
                                "  Consider registering with: noetl catalog register {}",
                                file_path.display()
                            );
                            // Use filename as path for now
                            let name = file_path
                                .file_stem()
                                .map(|s| s.to_string_lossy().to_string())
                                .unwrap_or_else(|| "playbook".to_string());
                            (name, None)
                        }
                        RefType::Catalog { name, version } => (name.clone(), version.clone()),
                        RefType::DatabaseId(id) => {
                            (id.clone(), None) // Server handles ID lookup
                        }
                        RefType::CatalogPath(path) => (path.clone(), exec_ctx.version.clone()),
                    };

                    let request_payload = build_distributed_payload(input.as_ref(), payload.as_deref(), &variables)?;

                    execute_playbook_distributed(
                        &client,
                        &effective_endpoint,
                        effective_gateway_proxy,
                        &path,
                        version.and_then(|v| v.parse().ok()),
                        request_payload,
                        ExecutionResourceKind::Playbook,
                        json,
                    )
                    .await?;
                }
                _ => {
                    eprintln!(
                        "Error: Unknown runtime '{}'. Use: local, distributed, or auto",
                        effective_runtime
                    );
                    std::process::exit(1);
                }
            }
        }
        Some(Commands::RunLegacy {
            reference,
            runtime,
            args,
            variables,
            payload,
            merge,
            verbose,
        }) => {
            // Get context runtime preference (honours --context override).
            let context_runtime = effective_context(context_override.as_deref(), &config).map(|(_, ctx)| ctx.runtime.as_str());

            // Determine effective runtime (context takes precedence for "auto")
            let effective_runtime = if runtime == "auto" {
                context_runtime.unwrap_or("local")
            } else {
                &runtime
            };

            // Build variables
            let vars = build_variables(payload.as_deref(), &variables)?;

            if effective_runtime == "distributed" {
                // Distributed execution via server
                let catalog_path = if let Some(ref_str) = &reference {
                    // Use the reference as catalog path
                    // Strip .yaml/.yml extension if present for catalog lookup
                    ref_str.trim_end_matches(".yaml").trim_end_matches(".yml").to_string()
                } else {
                    eprintln!("Error: Playbook reference required for distributed execution");
                    std::process::exit(1);
                };

                if verbose {
                    println!("Catalog path: {}", catalog_path);
                    println!("Runtime: {}", effective_runtime);
                    println!("Server: {}", base_url);
                }

                execute_playbook_distributed(
                    &client,
                    &base_url,
                    use_gateway_proxy,
                    &catalog_path,
                    None, // version
                    build_distributed_payload(None, payload.as_deref(), &variables)?,
                    ExecutionResourceKind::Playbook,
                    false, // json output
                )
                .await?;
            } else {
                // Local execution - resolve playbook file
                let (playbook_path, target) = if let Some(ref_str) = reference {
                    resolve_playbook_and_target(Some(ref_str), args)?
                } else {
                    resolve_playbook_and_target(None, args)?
                };

                if verbose {
                    println!("Resolved playbook: {}", playbook_path.display());
                    if let Some(ref t) = target {
                        println!("Target: {}", t);
                    }
                    println!("Runtime: {}", effective_runtime);
                }

                let runner = playbook_runner::PlaybookRunner::new(playbook_path)
                    .with_variables(vars)
                    .with_merge(merge)
                    .with_verbose(verbose)
                    .with_target(target);

                runner.run()?;
            }
        }
        Some(Commands::Context { command }) => {
            handle_context_command(&mut config, command).await?;
        }
        Some(Commands::Auth { command }) => {
            handle_auth_command(&mut config, &base_url, command).await?;
        }
        Some(Commands::Register { resource }) => match resource {
            RegisterResource::Credential { file, directory } => {
                if let Some(f) = file {
                    register_resource(&client, &base_url, use_gateway_proxy, "Credential", &f).await?;
                } else if let Some(d) = directory {
                    register_directory(&client, &base_url, use_gateway_proxy, "Credential", &d, &["json"]).await?;
                } else {
                    eprintln!("Error: Either --file or --directory must be specified");
                    std::process::exit(1);
                }
            }
            RegisterResource::Playbook { file, directory } => {
                if let Some(f) = file {
                    register_resource(&client, &base_url, use_gateway_proxy, "Playbook", &f).await?;
                } else if let Some(d) = directory {
                    register_directory(&client, &base_url, use_gateway_proxy, "Playbook", &d, &["yaml", "yml"]).await?;
                } else {
                    eprintln!("Error: Either --file or --directory must be specified");
                    std::process::exit(1);
                }
            }
        },
        Some(Commands::Status { execution_id, json }) => {
            get_status(&client, &base_url, use_gateway_proxy, &execution_id, json).await?;
        }
        Some(Commands::Cancel {
            execution_id,
            reason,
            cascade,
            json,
        }) => {
            cancel_execution(
                &client,
                &base_url,
                use_gateway_proxy,
                &execution_id,
                reason,
                cascade,
                json,
            )
            .await?;
        }
        Some(Commands::List { resource_type, json }) => {
            list_resources(&client, &base_url, use_gateway_proxy, &resource_type, json).await?;
        }
        Some(Commands::Catalog { command }) => {
            match command {
                CatalogCommand::Register { file } => {
                    // Auto-detect type from file content
                    let content =
                        fs::read_to_string(&file).context(format!("Failed to read file: {:?}", file.display()))?;
                    let resource_type = if content.contains("kind: Credential") {
                        "Credential"
                    } else if content.contains("kind: Playbook") {
                        "Playbook"
                    } else {
                        "Playbook" // Default
                    };
                    register_resource(&client, &base_url, use_gateway_proxy, resource_type, &file).await?;
                }
                CatalogCommand::Get { path } => {
                    get_catalog_resource(&client, &base_url, use_gateway_proxy, &path).await?;
                }
                CatalogCommand::List { resource_type, json } => {
                    list_resources(&client, &base_url, use_gateway_proxy, &resource_type, json).await?;
                }
            }
        }
        Some(Commands::Execute { command }) => match command {
            ExecuteCommand::Playbook { path, input, json } => {
                execute_playbook(&client, &base_url, use_gateway_proxy, &path, input, json).await?;
            }
            ExecuteCommand::Rerun {
                execution_id,
                input,
                json,
            } => {
                let workload = build_distributed_payload(input.as_ref(), None, &[])?;
                execute_rerun_distributed(
                    &client,
                    &base_url,
                    use_gateway_proxy,
                    &execution_id,
                    workload,
                    ExecutionResourceKind::Playbook,
                    json,
                )
                .await?;
            }
            ExecuteCommand::Status { execution_id, json } => {
                get_status(&client, &base_url, use_gateway_proxy, &execution_id, json).await?;
            }
        },
        Some(Commands::Get { resource }) => match resource {
            GetResource::Credential { name, include_data } => {
                get_credential(&client, &base_url, use_gateway_proxy, &name, include_data).await?;
            }
        },
        Some(Commands::Query { query, schema, format }) => {
            execute_query(&client, &base_url, use_gateway_proxy, &query, &schema, &format).await?;
        }
        Some(Commands::Server { command }) => match command {
            ServerCommand::Start { init_db } => {
                start_server(init_db).await?;
            }
            ServerCommand::Stop { force } => {
                stop_server(force).await?;
            }
        },
        Some(Commands::Worker { command }) => match command {
            WorkerCommand::Start { max_workers } => {
                start_worker(max_workers).await?;
            }
            WorkerCommand::Stop { name, force } => {
                stop_worker(name, force).await?;
            }
        },
        Some(Commands::Db { command }) => match command {
            DbCommand::Init => {
                db_init(&client, &base_url, use_gateway_proxy).await?;
            }
            DbCommand::Validate => {
                db_validate(&client, &base_url, use_gateway_proxy).await?;
            }
        },
        Some(Commands::Build { no_cache, platform }) => {
            build_docker_image(no_cache, platform).await?;
        }
        Some(Commands::K8s { command }) => match command {
            K8sCommand::Deploy => {
                k8s_deploy().await?;
            }
            K8sCommand::Remove => {
                k8s_remove().await?;
            }
            K8sCommand::Redeploy { no_cache, platform } => {
                k8s_redeploy(no_cache, platform).await?;
            }
            K8sCommand::Reset { no_cache, platform } => {
                k8s_reset(no_cache, platform).await?;
            }
        },
        Some(Commands::Cleanup {
            older_than_minutes,
            dry_run,
            json,
        }) => {
            cleanup_stuck_executions(&client, &base_url, use_gateway_proxy, older_than_minutes, dry_run, json).await?;
        }
        Some(Commands::Iap { command }) => {
            handle_iap_command(command).await?;
        }
        None => {
            println!("Use --help for usage information or --interactive for TUI mode");
        }
    }

    Ok(())
}

async fn handle_context_command(config: &mut Config, command: ContextCommand) -> Result<()> {
    match command {
        ContextCommand::Add {
            name,
            server_url,
            runtime,
            auth0_domain,
            auth0_client_id,
            auth0_redirect_uri,
            auth0_audience,
            auth0_client_secret,
            kube_context,
            kube_namespace,
            kube_service,
            kube_remote_port,
            set_current,
        } => {
            // Validate runtime value
            if !["local", "distributed", "auto"].contains(&runtime.as_str()) {
                eprintln!("Invalid runtime '{}'. Use: local, distributed, or auto", runtime);
                std::process::exit(1);
            }

            // Upsert behavior:
            // - Keep cached gateway session token for existing contexts.
            // - Update only provided Auth0 fields; omitted fields stay as-is.
            let mut context = config
                .contexts
                .get(&name)
                .cloned()
                .unwrap_or_else(|| Context::new(server_url.clone()));

            context.server_url = server_url;
            context.runtime = runtime;
            if let Some(domain) = auth0_domain {
                context.gateway_auth0_domain = Some(domain);
            }
            if let Some(client_id) = auth0_client_id {
                context.gateway_auth0_client_id = Some(client_id);
            }
            if let Some(redirect_uri) = auth0_redirect_uri {
                context.gateway_auth0_redirect_uri = Some(redirect_uri);
            }
            if let Some(audience) = auth0_audience {
                if audience.trim().is_empty() {
                    context.gateway_auth0_audience = None;
                } else {
                    context.gateway_auth0_audience = Some(audience);
                }
            }
            if let Some(client_secret) = auth0_client_secret {
                if client_secret.trim().is_empty() {
                    context.gateway_auth0_client_secret = None;
                } else {
                    context.gateway_auth0_client_secret = Some(client_secret);
                }
            }
            if let Some(value) = kube_context {
                context.kube_context =
                    if value.trim().is_empty() { None } else { Some(value) };
            }
            if let Some(value) = kube_namespace {
                context.kube_namespace =
                    if value.trim().is_empty() { None } else { Some(value) };
            }
            if let Some(value) = kube_service {
                context.kube_service =
                    if value.trim().is_empty() { None } else { Some(value) };
            }
            if let Some(port) = kube_remote_port {
                context.kube_remote_port = if port == 0 { None } else { Some(port) };
            }

            config.contexts.insert(name.clone(), context);
            if set_current || config.current_context.is_none() {
                config.current_context = Some(name.clone());
            }
            config.save()?;
            println!("Context '{}' added/updated.", name);
            if config.current_context.as_ref() == Some(&name) {
                println!("Context '{}' is now the current context.", name);
            }
        }
        ContextCommand::List => {
            println!(
                "  {:<15} {:<30} {:<12} {:<22} {:<10}",
                "NAME", "SERVER URL", "RUNTIME", "AUTH0 DOMAIN", "TOKEN"
            );
            for (name, ctx) in &config.contexts {
                let current_mark = if config.current_context.as_ref() == Some(name) {
                    "*"
                } else {
                    " "
                };
                let auth0_domain = ctx.gateway_auth0_domain.as_deref().unwrap_or("-");
                let token_state = if ctx.gateway_session_token.is_some() {
                    "cached"
                } else {
                    "-"
                };
                println!(
                    "{} {:<15} {:<30} {:<12} {:<22} {:<10}",
                    current_mark, name, ctx.server_url, ctx.runtime, auth0_domain, token_state
                );
            }
        }
        ContextCommand::Use { name } => {
            if config.contexts.contains_key(&name) {
                config.current_context = Some(name.clone());
                config.save()?;
                let ctx = config.contexts.get(&name).unwrap();
                println!("Switched to context '{}' (runtime: {}).", name, ctx.runtime);
            } else {
                eprintln!("Context '{}' not found.", name);
                std::process::exit(1);
            }
        }
        ContextCommand::SetRuntime { runtime } => {
            // Validate runtime value
            if !["local", "distributed", "auto"].contains(&runtime.as_str()) {
                eprintln!("Invalid runtime '{}'. Use: local, distributed, or auto", runtime);
                std::process::exit(1);
            }

            if let Some(name) = &config.current_context {
                if let Some(ctx) = config.contexts.get_mut(name) {
                    ctx.runtime = runtime.clone();
                    config.save()?;
                    println!("Runtime for context '{}' set to '{}'.", name, runtime);
                } else {
                    eprintln!("Current context '{}' not found in contexts.", name);
                    std::process::exit(1);
                }
            } else {
                eprintln!("No current context set. Use 'noetl context use <name>' first.");
                std::process::exit(1);
            }
        }
        ContextCommand::Init {
            name,
            from_gateway,
            non_interactive,
            runtime,
            set_current,
        } => {
            handle_context_init(config, &name, &from_gateway, &runtime, non_interactive, set_current)
                .await?;
        }
        ContextCommand::Update {
            name,
            server_url,
            runtime,
            auth0_domain,
            auth0_client_id,
            auth0_redirect_uri,
            auth0_audience,
            auth0_client_secret,
            kube_context,
            kube_namespace,
            kube_service,
            kube_remote_port,
        } => {
            // Update only patches fields that were explicitly passed.
            // It refuses to create a missing context — the operator
            // gets a clear hint to use ``context add`` instead.  This
            // is the friction case the round-01 handoff calls out:
            // an existing context with one missing Auth0 field can
            // be repaired with a single flag instead of being
            // re-created from scratch.
            let ctx = match config.contexts.get_mut(&name) {
                Some(c) => c,
                None => {
                    eprintln!(
                        "Context '{}' not found.  Use 'noetl context add {} --server-url=...' to create it.",
                        name, name
                    );
                    std::process::exit(1);
                }
            };

            if let Some(url) = server_url {
                let url = url.trim();
                if url.is_empty() {
                    eprintln!("--server-url cannot be empty.");
                    std::process::exit(1);
                }
                ctx.server_url = url.to_string();
            }
            if let Some(rt) = runtime {
                if !["local", "distributed", "auto"].contains(&rt.as_str()) {
                    eprintln!(
                        "Invalid runtime '{}'. Use: local, distributed, or auto",
                        rt
                    );
                    std::process::exit(1);
                }
                ctx.runtime = rt;
            }
            // Empty-string ⇒ clear the Auth0 field; non-empty ⇒ set
            // it.  Mirrors the clear-on-empty behaviour the existing
            // ``Add`` handler uses for audience + client_secret.
            if let Some(value) = auth0_domain {
                ctx.gateway_auth0_domain =
                    if value.trim().is_empty() { None } else { Some(value) };
            }
            if let Some(value) = auth0_client_id {
                ctx.gateway_auth0_client_id =
                    if value.trim().is_empty() { None } else { Some(value) };
            }
            if let Some(value) = auth0_redirect_uri {
                ctx.gateway_auth0_redirect_uri =
                    if value.trim().is_empty() { None } else { Some(value) };
            }
            if let Some(value) = auth0_audience {
                ctx.gateway_auth0_audience =
                    if value.trim().is_empty() { None } else { Some(value) };
            }
            if let Some(value) = auth0_client_secret {
                ctx.gateway_auth0_client_secret =
                    if value.trim().is_empty() { None } else { Some(value) };
            }
            // Kubernetes fields (paired with ``noetl context port-forward``).
            if let Some(value) = kube_context {
                ctx.kube_context =
                    if value.trim().is_empty() { None } else { Some(value) };
            }
            if let Some(value) = kube_namespace {
                ctx.kube_namespace =
                    if value.trim().is_empty() { None } else { Some(value) };
            }
            if let Some(value) = kube_service {
                ctx.kube_service =
                    if value.trim().is_empty() { None } else { Some(value) };
            }
            if let Some(port) = kube_remote_port {
                ctx.kube_remote_port = if port == 0 { None } else { Some(port) };
            }

            config.save()?;
            println!("Context '{}' updated.", name);
        }
        ContextCommand::Delete { name } => {
            if config.contexts.remove(&name).is_some() {
                if config.current_context.as_ref() == Some(&name) {
                    config.current_context = None;
                }
                config.save()?;
                println!("Context '{}' deleted.", name);
            } else {
                eprintln!("Context '{}' not found.", name);
                std::process::exit(1);
            }
        }
        ContextCommand::Current => {
            if let Some((name, ctx)) = config.get_current_context() {
                println!("Current context: {}", name);
                println!("  Server URL: {}", ctx.server_url);
                println!("  Runtime:    {}", ctx.runtime);
                println!(
                    "  Auth0:      {}",
                    ctx.gateway_auth0_domain.as_deref().unwrap_or("(gateway default)")
                );
                println!(
                    "  Audience:   {}",
                    ctx.gateway_auth0_audience.as_deref().unwrap_or("(none)")
                );
                println!(
                    "  Token:      {}",
                    if ctx.gateway_session_token.is_some() {
                        "cached"
                    } else {
                        "not cached"
                    }
                );
                if let Some(kc) = &ctx.kube_context {
                    println!("  Kube ctx:   {}", kc);
                }
                if let Some(ns) = &ctx.kube_namespace {
                    println!("  Kube ns:    {}", ns);
                }
            } else {
                println!("No current context set.");
            }
        }
        ContextCommand::PortForward {
            name,
            detach,
            stop,
            status,
        } => {
            handle_port_forward(config, &name, detach, stop, status)?;
        }
    }
    Ok(())
}

/// PID-file location for a context's detached ``kubectl port-forward``.
/// ``~/.config/noetl/port-forwards/<context>.pid``.  The directory is
/// created lazily on first ``--detach``.
fn port_forward_pid_path(context_name: &str) -> Result<PathBuf> {
    let home = dirs::home_dir().context("Could not find home directory")?;
    let dir = home.join(".noetl").join("port-forwards");
    Ok(dir.join(format!("{}.pid", context_name)))
}

/// Effective port-forward target for a context.  Returns
/// ``(kube_context, kube_namespace, service, local_port, remote_port)``
/// or ``None`` if the context lacks the required ``kube_context`` /
/// ``kube_namespace`` fields.  ``service`` defaults to ``noetl`` and
/// ``remote_port`` defaults to 8082 when unset.  ``local_port`` is
/// derived from the context's ``server_url`` (we forward to whatever
/// port the operator already configured the CLI to call).
fn port_forward_target(ctx: &Context) -> Option<(String, String, String, u16, u16)> {
    let kube_context = ctx.kube_context.clone()?;
    let kube_namespace = ctx.kube_namespace.clone()?;
    let service = ctx
        .kube_service
        .clone()
        .unwrap_or_else(|| "noetl".to_string());
    let remote_port = ctx.kube_remote_port.unwrap_or(8082);
    let local_port = parse_local_port_from_server_url(&ctx.server_url)?;
    Some((kube_context, kube_namespace, service, local_port, remote_port))
}

/// Extract the localhost port from a server URL like
/// ``http://127.0.0.1:18082`` or ``http://localhost:8082``.  Returns
/// ``None`` for non-localhost URLs — those don't make sense as
/// port-forward targets.
fn parse_local_port_from_server_url(url: &str) -> Option<u16> {
    let parsed = Url::parse(url).ok()?;
    let host = parsed.host_str()?;
    if !matches!(host, "localhost" | "127.0.0.1" | "::1") {
        return None;
    }
    parsed.port()
}

/// Pure helper: parse the ``auth0`` block from a runtime contract
/// response into ``(domain, client_id, redirect_uri, audience)``.
/// Returns ``None`` when no usable ``auth0`` block is present
/// (older gateway image, or deployment without Auth0).  Empty
/// fields are normalised to empty strings — the caller decides
/// whether to clear or skip each one.
///
/// Split from the HTTP-fetching wrapper so it can be unit-tested
/// without a live gateway.
fn parse_auth0_block_from_contract(
    contract: &serde_json::Value,
) -> Option<(String, String, String, String)> {
    let auth0 = contract.get("auth0")?.as_object()?;
    let domain = auth0
        .get("domain")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .trim()
        .to_string();
    if domain.is_empty() {
        return None;
    }
    let client_id = auth0
        .get("client_id")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .trim()
        .to_string();
    let redirect_uri = auth0
        .get("redirect_uri")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .trim()
        .to_string();
    let audience = auth0
        .get("audience")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .trim()
        .to_string();
    Some((domain, client_id, redirect_uri, audience))
}

/// Normalise the gateway URL so the runtime-contract fetch and the
/// stored ``server_url`` are consistent: trim whitespace, drop a
/// trailing slash.  Returns an owned ``String``.
fn normalise_gateway_url(url: &str) -> String {
    url.trim().trim_end_matches('/').to_string()
}

async fn handle_context_init(
    config: &mut Config,
    name: &str,
    from_gateway: &str,
    runtime: &str,
    non_interactive: bool,
    set_current: bool,
) -> Result<()> {
    if !["local", "distributed", "auto"].contains(&runtime) {
        eprintln!("Invalid runtime '{}'. Use: local, distributed, or auto", runtime);
        std::process::exit(1);
    }

    let gateway_url = normalise_gateway_url(from_gateway);
    if gateway_url.is_empty() {
        eprintln!("--from-gateway cannot be empty.");
        std::process::exit(1);
    }
    let contract_url = format!("{}/api/runtime/contract", gateway_url);

    println!("Fetching runtime contract from {} ...", contract_url);
    let client = reqwest::Client::new();
    let response = client
        .get(&contract_url)
        .send()
        .await
        .with_context(|| format!("Failed to GET {}", contract_url))?;

    if !response.status().is_success() {
        eprintln!(
            "Gateway returned {} from {}.  Cannot bootstrap context.",
            response.status(),
            contract_url
        );
        std::process::exit(1);
    }

    let contract: serde_json::Value = response
        .json()
        .await
        .context("Failed to parse runtime contract as JSON")?;

    let auth0 = parse_auth0_block_from_contract(&contract);

    // Show the operator what we're about to write before we touch
    // the config.  Even with ``--yes`` the table prints — it doubles
    // as the post-hoc "what got written" log line.
    println!();
    println!("About to write context '{}':", name);
    println!("  server_url:           {}", gateway_url);
    println!("  runtime:              {}", runtime);
    if let Some((domain, client_id, redirect_uri, audience)) = auth0.clone() {
        println!("  auth0_domain:         {}", domain);
        println!(
            "  auth0_client_id:      {}",
            if client_id.is_empty() { "(none)" } else { &client_id }
        );
        println!(
            "  auth0_redirect_uri:   {}",
            if redirect_uri.is_empty() { "(none)" } else { &redirect_uri }
        );
        println!(
            "  auth0_audience:       {}",
            if audience.is_empty() { "(none)" } else { &audience }
        );
    } else {
        println!("  auth0:                (not exposed by this gateway — see warning below)");
    }
    if set_current {
        println!("  set as current:       yes");
    }
    println!();

    if !non_interactive {
        print!("Proceed? [y/N] ");
        use std::io::Write;
        std::io::stdout().flush().ok();
        let mut answer = String::new();
        std::io::stdin()
            .read_line(&mut answer)
            .context("Failed to read confirmation from stdin")?;
        let answer = answer.trim().to_lowercase();
        if !matches!(answer.as_str(), "y" | "yes") {
            println!("Aborted.  No context changes written.");
            return Ok(());
        }
    }

    if auth0.is_none() {
        eprintln!(
            "Warning: gateway runtime contract does not carry an ``auth0`` block."
        );
        eprintln!(
            "  Either the gateway image is older than v2.10 (no Auth0 exposure)"
        );
        eprintln!("  or the deployment is configured without Auth0.  Context will be");
        eprintln!("  created with ``server_url`` only.  Add Auth0 fields manually if needed:");
        eprintln!(
            "    noetl context update {} --auth0-domain=... --auth0-client-id=...",
            name
        );
    }

    // Apply directly — Init replaces any existing context with the
    // same name (it's a bootstrap command; the operator's intent is
    // a clean slate from the gateway's perspective).
    let mut context = Context::new(gateway_url.clone());
    context.runtime = runtime.to_string();
    if let Some((domain, client_id, redirect_uri, audience)) = auth0 {
        context.gateway_auth0_domain = Some(domain);
        if !client_id.is_empty() {
            context.gateway_auth0_client_id = Some(client_id);
        }
        if !redirect_uri.is_empty() {
            context.gateway_auth0_redirect_uri = Some(redirect_uri);
        }
        if !audience.is_empty() {
            context.gateway_auth0_audience = Some(audience);
        }
    }

    config.contexts.insert(name.to_string(), context);
    if set_current {
        config.current_context = Some(name.to_string());
    }
    config.save()?;
    println!("Context '{}' written.", name);
    if set_current {
        println!("Context '{}' is now the current context.", name);
    }
    println!();
    println!("Next step:  noetl auth login --browser-pkce --context {}", name);
    Ok(())
}

fn handle_port_forward(
    config: &Config,
    name: &str,
    detach: bool,
    stop: bool,
    status: bool,
) -> Result<()> {
    let ctx = match config.contexts.get(name) {
        Some(c) => c,
        None => {
            eprintln!("Context '{}' not found.", name);
            std::process::exit(1);
        }
    };

    let pid_path = port_forward_pid_path(name)?;

    if status {
        match read_pid_file(&pid_path)? {
            Some(pid) if pid_is_running(pid) => {
                println!("Port-forward for context '{}' running, pid {}.", name, pid);
            }
            Some(pid) => {
                println!(
                    "Port-forward for context '{}' has stale pid file (pid {} not running).",
                    name, pid
                );
                println!("  PID file: {}", pid_path.display());
                println!("  Remove with: noetl context port-forward {} --stop", name);
            }
            None => {
                println!("No detached port-forward registered for context '{}'.", name);
            }
        }
        return Ok(());
    }

    if stop {
        match read_pid_file(&pid_path)? {
            Some(pid) => {
                if pid_is_running(pid) {
                    kill_pid_with_grace(pid);
                    println!("Stopped port-forward for context '{}' (pid {}).", name, pid);
                } else {
                    println!(
                        "Port-forward for context '{}' was not running (pid {} dead).",
                        name, pid
                    );
                }
                let _ = std::fs::remove_file(&pid_path);
            }
            None => {
                println!("No detached port-forward registered for context '{}'.", name);
            }
        }
        return Ok(());
    }

    // start (foreground or detached)
    let (kube_ctx, kube_ns, service, local_port, remote_port) = match port_forward_target(ctx) {
        Some(t) => t,
        None => {
            eprintln!(
                "Context '{}' is missing port-forward configuration.  Set the kube fields first:",
                name
            );
            eprintln!();
            eprintln!(
                "  noetl context update {} \\\\\n      --kube-context=<gke-or-kind-context> \\\\\n      --kube-namespace=<namespace> \\\\\n      [--kube-service=<service-name>] \\\\\n      [--kube-remote-port=<in-cluster-port>]",
                name
            );
            eprintln!();
            eprintln!(
                "The local port is derived from the context's ``server_url`` (must be a"
            );
            eprintln!(
                "loopback URL like http://127.0.0.1:18082 or http://localhost:8082)."
            );
            std::process::exit(1);
        }
    };

    // Refuse to start a duplicate detached forward (our own PID file
    // says one is alive).
    if detach {
        if let Some(existing) = read_pid_file(&pid_path)? {
            if pid_is_running(existing) {
                eprintln!(
                    "A detached port-forward for context '{}' is already running (pid {}).",
                    name, existing
                );
                eprintln!("  Stop it first:  noetl context port-forward {} --stop", name);
                std::process::exit(1);
            }
            // Stale pid file — clean it up and continue.
            let _ = std::fs::remove_file(&pid_path);
        }
    }

    // Refuse to start the kubectl tunnel when the local port is
    // already bound by ANY process (the existing PID-file check
    // above only catches *our own* prior daemons).  Without this,
    // a stray kubectl from a different session would happily let
    // ``--detach`` "succeed" — the new kubectl spawns, sees the
    // port-in-use error a few hundred ms later, and exits 1,
    // leaving an orphan PID file pointing at a dead process.
    // ``--status`` then reports "stale pid file" with no hint why.
    //
    // Probe by attempting to bind a TcpListener to the same loopback
    // port.  If bind succeeds, we drop the listener immediately —
    // there's a tiny race window between this check and kubectl
    // claiming the port, but it's vastly better than no check at
    // all.  We probe ``127.0.0.1`` since that's what kubectl binds.
    {
        use std::net::TcpListener;
        let probe_addr = format!("127.0.0.1:{}", local_port);
        match TcpListener::bind(&probe_addr) {
            Ok(_) => {
                // Free.  Listener drops here, releasing the port.
            }
            Err(err) => {
                eprintln!(
                    "Local port {} is already in use (bind probe failed: {}).",
                    local_port, err
                );
                eprintln!("  Identify the process:");
                eprintln!("    lsof -i :{}                 # macOS / Linux", local_port);
                eprintln!("    fuser {}/tcp                # Linux", local_port);
                eprintln!();
                eprintln!("  If it's a stray ``kubectl port-forward`` from another shell:");
                eprintln!("    pkill -f 'port-forward svc/noetl'");
                eprintln!();
                eprintln!(
                    "  If it's another noetl-managed daemon: ``noetl context port-forward <other> --stop``."
                );
                std::process::exit(1);
            }
        }
    }

    let local_port_str = local_port.to_string();
    let port_arg = format!("{}:{}", local_port, remote_port);
    let svc_arg = format!("svc/{}", service);
    let kubectl_args = vec![
        "--context".to_string(),
        kube_ctx.clone(),
        "-n".to_string(),
        kube_ns.clone(),
        "port-forward".to_string(),
        svc_arg.clone(),
        port_arg.clone(),
    ];

    if !detach {
        println!(
            "Starting kubectl port-forward (foreground) for context '{}':",
            name
        );
        println!(
            "  kubectl --context {} -n {} port-forward {} {}",
            kube_ctx, kube_ns, svc_arg, port_arg
        );
        println!("  Local port: {}", local_port_str);
        println!("  Press Ctrl-C to stop.");
        let status = std::process::Command::new("kubectl")
            .args(&kubectl_args)
            .status()
            .context("Failed to spawn kubectl.  Is it installed and on PATH?")?;
        if !status.success() {
            std::process::exit(status.code().unwrap_or(1));
        }
        return Ok(());
    }

    // Detached path — fork via std::process::Command with stdio detached.
    if let Some(parent) = pid_path.parent() {
        std::fs::create_dir_all(parent).with_context(|| {
            format!("Failed to create port-forward PID directory: {}", parent.display())
        })?;
    }
    let child = std::process::Command::new("kubectl")
        .args(&kubectl_args)
        .stdin(std::process::Stdio::null())
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::null())
        .spawn()
        .context("Failed to spawn kubectl port-forward as daemon")?;
    let pid = child.id();
    std::fs::write(&pid_path, pid.to_string())
        .with_context(|| format!("Failed to write PID file: {}", pid_path.display()))?;

    // Quick sanity check — give kubectl ~250ms to fail fast (e.g. bind
    // error from a port already in use).  If it died, surface that.
    std::thread::sleep(std::time::Duration::from_millis(250));
    if !pid_is_running(pid) {
        let _ = std::fs::remove_file(&pid_path);
        eprintln!(
            "kubectl port-forward exited immediately (pid {} no longer running).",
            pid
        );
        eprintln!(
            "  Common causes: local port {} already in use, kube_context not in kubeconfig, namespace/service not found.",
            local_port_str
        );
        eprintln!(
            "  Try running in foreground for full error output: noetl context port-forward {}",
            name
        );
        std::process::exit(1);
    }

    println!(
        "Detached kubectl port-forward for context '{}' (pid {}).",
        name, pid
    );
    println!("  Local:  http://127.0.0.1:{}", local_port_str);
    println!("  Remote: {}/{} port {}", kube_ns, service, remote_port);
    println!("  PID:    {}", pid_path.display());
    println!("  Stop:   noetl context port-forward {} --stop", name);
    Ok(())
}

fn read_pid_file(path: &PathBuf) -> Result<Option<u32>> {
    if !path.exists() {
        return Ok(None);
    }
    let content = std::fs::read_to_string(path)
        .with_context(|| format!("Failed to read PID file: {}", path.display()))?;
    let pid: u32 = content
        .trim()
        .parse()
        .with_context(|| format!("PID file is corrupt: {}", path.display()))?;
    Ok(Some(pid))
}

/// POSIX ``kill -0`` style liveness probe via the ``nix`` crate
/// (already a CLI dependency).  On non-Unix platforms this is a
/// best-effort: we can't reliably detect dead processes, so we
/// return ``true`` and let the caller's spawn-or-stop logic
/// recover from the resulting error.
#[cfg(unix)]
fn pid_is_running(pid: u32) -> bool {
    use nix::sys::signal::kill;
    use nix::unistd::Pid;
    kill(Pid::from_raw(pid as i32), None).is_ok()
}

#[cfg(not(unix))]
fn pid_is_running(_pid: u32) -> bool {
    // Best-effort fallback for non-Unix platforms.  The detached
    // ``noetl context port-forward`` path is documented as
    // Unix-only in the README; this branch keeps the code
    // compilable on Windows for testing the rest of the CLI.
    true
}

/// Send SIGTERM and, after a short grace, SIGKILL.  POSIX-only —
/// the daemon path requires Unix per the same caveat as
/// ``pid_is_running``.
#[cfg(unix)]
fn kill_pid_with_grace(pid: u32) {
    use nix::sys::signal::{kill, Signal};
    use nix::unistd::Pid;
    let target = Pid::from_raw(pid as i32);
    let _ = kill(target, Signal::SIGTERM);
    std::thread::sleep(std::time::Duration::from_millis(500));
    if pid_is_running(pid) {
        let _ = kill(target, Signal::SIGKILL);
    }
}

#[cfg(not(unix))]
fn kill_pid_with_grace(_pid: u32) {
    // No-op on non-Unix — see ``pid_is_running`` for the rationale.
}

fn extract_token_from_callback_url(url: &str) -> Option<String> {
    for source in [url.split('#').nth(1), url.split('?').nth(1)] {
        let Some(source) = source else { continue };
        let mut id_token: Option<String> = None;
        let mut access_token: Option<String> = None;
        for pair in source.split('&') {
            let mut parts = pair.splitn(2, '=');
            let key = match parts.next() {
                Some(k) => k.trim(),
                None => continue,
            };
            let value = parts.next().unwrap_or("").trim();
            if !value.is_empty() {
                match key {
                    "id_token" => id_token = Some(value.to_string()),
                    "access_token" => access_token = Some(value.to_string()),
                    _ => {}
                }
            }
        }
        // Always prefer id_token (JWT) over access_token (may be opaque)
        if let Some(t) = id_token.or(access_token) {
            return Some(t);
        }
    }
    None
}

fn looks_like_email(value: &str) -> bool {
    let trimmed = value.trim();
    trimmed.contains('@')
        && trimmed.split('@').count() == 2
        && trimmed.split('@').nth(1).is_some_and(|domain| domain.contains('.'))
}

fn looks_like_jwt(value: &str) -> bool {
    let trimmed = value.trim();
    trimmed.split('.').count() == 3 && !trimmed.contains(' ')
}

fn prompt_input(prompt: &str) -> Result<String> {
    use std::io::Write;
    print!("{}", prompt);
    io::stdout().flush().context("Failed to flush prompt to stdout")?;
    let mut line = String::new();
    io::stdin()
        .read_line(&mut line)
        .context("Failed to read input from stdin")?;
    Ok(line.trim().to_string())
}

fn mask_token(token: &str) -> String {
    if token.len() <= 12 {
        return "[redacted]".to_string();
    }
    format!("{}...{}", &token[..6], &token[token.len() - 4..])
}

fn open_browser(url: &str) -> Result<()> {
    if cfg!(target_os = "macos") {
        Command::new("open")
            .arg(url)
            .spawn()
            .context("Failed to launch browser with 'open'")?;
        Ok(())
    } else if cfg!(target_os = "linux") {
        Command::new("xdg-open")
            .arg(url)
            .spawn()
            .context("Failed to launch browser with 'xdg-open'")?;
        Ok(())
    } else if cfg!(target_os = "windows") {
        Command::new("cmd")
            .args(["/C", "start", "", url])
            .spawn()
            .context("Failed to launch browser with 'start'")?;
        Ok(())
    } else {
        anyhow::bail!("Automatic browser launch is not supported on this platform.")
    }
}

fn generate_random_urlsafe(bytes_len: usize) -> String {
    let mut bytes = vec![0_u8; bytes_len];
    OsRng.fill_bytes(&mut bytes);
    BASE64_URL_SAFE_NO_PAD.encode(bytes)
}

fn build_pkce_code_verifier() -> String {
    // RFC 7636: verifier length must be between 43 and 128 chars.
    generate_random_urlsafe(64)
}

fn build_pkce_code_challenge(verifier: &str) -> String {
    let digest = Sha256::digest(verifier.as_bytes());
    BASE64_URL_SAFE_NO_PAD.encode(digest)
}

fn parse_loopback_redirect_uri(redirect_uri: &str) -> Result<Url> {
    let redirect =
        Url::parse(redirect_uri.trim()).with_context(|| format!("Invalid redirect URI: {}", redirect_uri.trim()))?;
    let host = redirect
        .host_str()
        .ok_or_else(|| anyhow::anyhow!("Redirect URI must include host: {}", redirect_uri.trim()))?;
    let is_loopback = host.eq_ignore_ascii_case("localhost") || host == "127.0.0.1";
    if !is_loopback {
        anyhow::bail!(
            "Redirect URI host must be localhost or 127.0.0.1 for PKCE callback flow: {}",
            redirect_uri.trim()
        );
    }
    if redirect.port().is_none() {
        anyhow::bail!(
            "Redirect URI must include an explicit port for PKCE callback flow: {}",
            redirect_uri.trim()
        );
    }
    Ok(redirect)
}

fn resolve_pkce_redirect_uri(
    explicit_redirect_uri: Option<&str>,
    context_redirect_uri: Option<&str>,
    pkce_port: u16,
) -> Result<String> {
    if let Some(uri) = explicit_redirect_uri {
        let trimmed = uri.trim();
        if !trimmed.is_empty() {
            parse_loopback_redirect_uri(trimmed)?;
            return Ok(trimmed.to_string());
        }
    }

    if let Some(uri) = context_redirect_uri {
        let trimmed = uri.trim();
        if !trimmed.is_empty() && parse_loopback_redirect_uri(trimmed).is_ok() {
            return Ok(trimmed.to_string());
        }
    }

    Ok(format!("http://127.0.0.1:{}/callback", pkce_port))
}

async fn write_callback_http_response(
    socket: &mut tokio::net::TcpStream,
    status_line: &str,
    title: &str,
    message: &str,
) -> Result<()> {
    let body = format!(
        "<!doctype html><html><head><meta charset=\"utf-8\"><title>{}</title></head><body><h1>{}</h1><p>{}</p></body></html>",
        title, title, message
    );
    let response = format!(
        "HTTP/1.1 {}\r\nContent-Type: text/html; charset=utf-8\r\nCache-Control: no-store\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
        status_line,
        body.as_bytes().len(),
        body
    );
    socket
        .write_all(response.as_bytes())
        .await
        .context("Failed to write callback HTTP response")?;
    Ok(())
}

/// Extract the Auth0 tenant slug from an Auth0 domain like
/// ``mestumre-development.us.auth0.com`` ⇒ ``mestumre-development``.
/// Used (with the optional region label) to construct the dashboard
/// management URL the user clicks when the PKCE callback URL isn't
/// in the application's allowed list.  Returns ``None`` for custom
/// domains (CNAMEd) where the tenant slug can't be derived from the
/// domain alone.
fn extract_auth0_tenant_from_domain(domain: &str) -> Option<String> {
    extract_auth0_tenant_and_region(domain).map(|(tenant, _)| tenant)
}

/// Extract both the tenant slug and the optional region label from
/// an Auth0 domain.  Standard Auth0 domain shapes:
///
/// - ``<tenant>.auth0.com``               ⇒ ``(tenant, None)``
/// - ``<tenant>.<region>.auth0.com``      ⇒ ``(tenant, Some(region))``
///   where ``region`` is one of ``us`` / ``eu`` / ``au`` / ``jp``.
///
/// Auth0's management dashboard URL embeds the region for regional
/// tenants, so callers building dashboard URLs need both pieces.
/// See [``build_auth0_dashboard_url``].
fn extract_auth0_tenant_and_region(domain: &str) -> Option<(String, Option<String>)> {
    let trimmed = domain.trim().trim_end_matches('/').to_lowercase();
    if !trimmed.ends_with(".auth0.com") {
        return None;
    }
    let labels: Vec<&str> = trimmed.split('.').collect();
    // Layout is at least [tenant, "auth0", "com"]; with a region it's
    // [tenant, region, "auth0", "com"].  Anything else (e.g. extra
    // sub-domains we don't recognise) returns None — better to fall
    // back to the ``<tenant>`` placeholder than to construct a wrong
    // URL.
    match labels.as_slice() {
        [tenant, "auth0", "com"] => Some(((*tenant).to_string(), None)),
        [tenant, region, "auth0", "com"]
            if matches!(*region, "us" | "eu" | "au" | "jp") =>
        {
            Some(((*tenant).to_string(), Some((*region).to_string())))
        }
        _ => None,
    }
}

/// Build the Auth0 management dashboard URL for a given application
/// in a tenant.  Auth0's dashboard URL format is:
///
/// - Default region (US, no region label in domain):
///   ``https://manage.auth0.com/dashboard/<tenant>/applications/<client_id>/settings``
/// - Regional tenants (``us`` / ``eu`` / ``au`` / ``jp`` label in
///   domain):
///   ``https://manage.auth0.com/dashboard/<region>/<tenant>/applications/<client_id>/settings``
///
/// Returns ``None`` when the domain isn't a recognised Auth0 shape
/// (e.g. CNAMEd custom domains) — callers should fall back to a
/// ``<tenant>`` placeholder in their hint output.
fn build_auth0_dashboard_url(domain: &str, client_id: &str) -> Option<String> {
    let (tenant, region) = extract_auth0_tenant_and_region(domain)?;
    let client_id_trimmed = client_id.trim();
    let url = match region {
        Some(region) => format!(
            "https://manage.auth0.com/dashboard/{}/{}/applications/{}/settings",
            region, tenant, client_id_trimmed
        ),
        None => format!(
            "https://manage.auth0.com/dashboard/{}/applications/{}/settings",
            tenant, client_id_trimmed
        ),
    };
    Some(url)
}

/// Wrapper around ``wait_for_pkce_callback`` that turns the bare
/// "Timed out waiting for Auth0 PKCE callback" error into a hint
/// pointing at the Auth0 application's allowed-callback-URLs page.
/// When Auth0 rejects the ``/authorize`` request because the
/// redirect_uri isn't in the application's allowed list, the user
/// sees a "Callback URL mismatch" page in their browser but the
/// CLI never gets a callback — it just sits at the listener until
/// the 5-minute timeout.  This is the single most common failure
/// shape for a fresh PKCE setup, so the timeout message names it
/// directly.
async fn wait_for_pkce_callback_with_hint(
    listener: &TcpListener,
    expected_path: &str,
    expected_state: &str,
    redirect_uri: &str,
    client_id: &str,
    domain: &str,
) -> Result<String> {
    match wait_for_pkce_callback(listener, expected_path, expected_state).await {
        Ok(code) => Ok(code),
        Err(err) => {
            // The original error message is preserved as the
            // ``anyhow`` chain root for log surfaces that want it,
            // but the human-facing hint is prepended to the
            // displayed bail message.
            let dashboard_url = build_auth0_dashboard_url(domain, client_id)
                .unwrap_or_else(|| {
                    format!(
                        "https://manage.auth0.com/dashboard/<tenant>/applications/{}/settings",
                        client_id.trim()
                    )
                });
            eprintln!();
            eprintln!(
                "PKCE callback did not arrive in time.  The most common cause is that"
            );
            eprintln!(
                "Auth0 rejected the /authorize request because the redirect URI"
            );
            eprintln!("isn't in the application's allowed callbacks list:");
            eprintln!();
            eprintln!("  redirect_uri: {}", redirect_uri);
            eprintln!();
            eprintln!("Open the Auth0 application settings and add the URI above to");
            eprintln!("\"Allowed Callback URLs\":");
            eprintln!();
            eprintln!("  {}", dashboard_url);
            eprintln!();
            eprintln!(
                "Then retry the login.  If the URI was already allowed, the timeout"
            );
            eprintln!(
                "may indicate a different failure (network, browser blocked the"
            );
            eprintln!("listener, or the Auth0 tenant/client_id is wrong).  Original error:");
            eprintln!("  {}", err);
            Err(err)
        }
    }
}

async fn wait_for_pkce_callback(listener: &TcpListener, expected_path: &str, expected_state: &str) -> Result<String> {
    let timeout_at = Instant::now() + Duration::from_secs(300);

    loop {
        let now = Instant::now();
        if now >= timeout_at {
            anyhow::bail!("Timed out waiting for Auth0 PKCE callback.");
        }
        let remaining = timeout_at.saturating_duration_since(now);

        let (mut socket, _) = tokio::time::timeout(remaining, listener.accept())
            .await
            .context("Timed out waiting for callback connection")?
            .context("Failed to accept callback connection")?;

        let mut request_buf = vec![0_u8; 8192];
        let bytes_read = tokio::time::timeout(Duration::from_secs(15), socket.read(&mut request_buf))
            .await
            .context("Timed out reading callback request")?
            .context("Failed to read callback request")?;
        if bytes_read == 0 {
            continue;
        }

        let request_line = String::from_utf8_lossy(&request_buf[..bytes_read])
            .lines()
            .next()
            .unwrap_or_default()
            .to_string();
        let mut parts = request_line.split_whitespace();
        let method = parts.next().unwrap_or_default();
        let target = parts.next().unwrap_or_default();

        if method != "GET" || target.is_empty() {
            let _ = write_callback_http_response(
                &mut socket,
                "405 Method Not Allowed",
                "NoETL Login",
                "Invalid callback request method.",
            )
            .await;
            continue;
        }

        let callback_url = match Url::parse(&format!("http://localhost{}", target)) {
            Ok(url) => url,
            Err(_) => {
                let _ = write_callback_http_response(
                    &mut socket,
                    "400 Bad Request",
                    "NoETL Login",
                    "Failed to parse callback URL.",
                )
                .await;
                continue;
            }
        };

        if callback_url.path() != expected_path {
            let _ =
                write_callback_http_response(&mut socket, "404 Not Found", "NoETL Login", "Unexpected callback path.")
                    .await;
            continue;
        }

        let params: HashMap<String, String> = callback_url.query_pairs().into_owned().collect();
        if let Some(error) = params.get("error") {
            let _ = write_callback_http_response(
                &mut socket,
                "400 Bad Request",
                "NoETL Login Failed",
                "Authentication failed. Return to the terminal for details.",
            )
            .await;
            let description = params.get("error_description").cloned().unwrap_or_default();
            anyhow::bail!("Auth0 callback error: {} {}", error, description);
        }

        let returned_state = params.get("state").map(String::as_str).unwrap_or_default();
        if returned_state != expected_state {
            let _ = write_callback_http_response(
                &mut socket,
                "400 Bad Request",
                "NoETL Login Failed",
                "State verification failed. Return to the terminal and retry.",
            )
            .await;
            anyhow::bail!("PKCE callback state mismatch.");
        }

        let code = params
            .get("code")
            .filter(|value| !value.trim().is_empty())
            .cloned()
            .ok_or_else(|| anyhow::anyhow!("Auth0 callback missing authorization code"))?;

        write_callback_http_response(
            &mut socket,
            "200 OK",
            "NoETL Login Successful",
            "Authentication complete. You can close this window and return to the terminal.",
        )
        .await?;

        return Ok(code);
    }
}

async fn auth0_pkce_authorize(
    domain: &str,
    client_id: &str,
    redirect_uri: &str,
    audience: Option<&str>,
    client_secret: Option<&str>,
    login_hint: Option<&str>,
    open: bool,
    json: bool,
) -> Result<String> {
    let redirect = parse_loopback_redirect_uri(redirect_uri)?;
    let host = redirect.host_str().unwrap_or("127.0.0.1");
    let bind_host = if host.eq_ignore_ascii_case("localhost") {
        "127.0.0.1"
    } else {
        host
    };
    let port = redirect
        .port()
        .ok_or_else(|| anyhow::anyhow!("Redirect URI must include an explicit port"))?;
    let listener_addr = format!("{}:{}", bind_host, port);
    let listener = TcpListener::bind(&listener_addr)
        .await
        .with_context(|| format!("Failed to bind local callback listener at {}", listener_addr))?;

    let verifier = build_pkce_code_verifier();
    let challenge = build_pkce_code_challenge(&verifier);
    let state = generate_random_urlsafe(24);
    let token_url = format!("https://{}/oauth/token", domain.trim());

    let mut authorize_url =
        Url::parse(&format!("https://{}/authorize", domain.trim())).context("Invalid Auth0 domain")?;
    {
        let mut query = authorize_url.query_pairs_mut();
        query.append_pair("response_type", "code");
        query.append_pair("client_id", client_id.trim());
        query.append_pair("redirect_uri", redirect_uri.trim());
        query.append_pair("scope", "openid profile email");
        query.append_pair("code_challenge", &challenge);
        query.append_pair("code_challenge_method", "S256");
        query.append_pair("state", &state);
        if let Some(aud) = audience {
            if !aud.trim().is_empty() {
                query.append_pair("audience", aud.trim());
            }
        }
        if let Some(hint) = login_hint {
            if !hint.trim().is_empty() {
                query.append_pair("login_hint", hint.trim());
            }
        }
    }
    let authorize_url = authorize_url.to_string();

    if !json {
        println!("PKCE callback listener ready at {}", listener_addr);
        println!("Redirect URI: {}", redirect_uri);
        println!(
            "  NOTE: this URI must appear in the Auth0 application's"
        );
        println!("        \"Allowed Callback URLs\" list, otherwise Auth0");
        println!("        will reject the /authorize request with a");
        println!("        \"Callback URL mismatch\" page and this CLI");
        println!("        will hang until timeout.  Manage Allowed");
        println!("        Callback URLs at:");
        println!(
            "        {}",
            build_auth0_dashboard_url(domain, client_id).unwrap_or_else(|| format!(
                "https://manage.auth0.com/dashboard/<tenant>/applications/{}/settings",
                client_id.trim()
            ))
        );
        println!("Open this URL to authenticate:");
        println!("  {}", authorize_url);
        if open {
            match open_browser(&authorize_url) {
                Ok(()) => println!("Opened browser for authentication."),
                Err(err) => eprintln!("Could not open browser automatically: {}", err),
            }
        }
        println!("Waiting for Auth0 callback...");
    }

    let code = wait_for_pkce_callback_with_hint(
        &listener,
        redirect.path(),
        &state,
        redirect_uri,
        client_id,
        domain,
    )
    .await?;

    let http_client = Client::new();
    let mut form: Vec<(String, String)> = vec![
        ("grant_type".to_string(), "authorization_code".to_string()),
        ("client_id".to_string(), client_id.trim().to_string()),
        ("code".to_string(), code),
        ("code_verifier".to_string(), verifier),
        ("redirect_uri".to_string(), redirect_uri.trim().to_string()),
    ];
    if let Some(secret) = client_secret {
        if !secret.trim().is_empty() {
            form.push(("client_secret".to_string(), secret.trim().to_string()));
        }
    }

    let response = http_client
        .post(&token_url)
        .form(&form)
        .send()
        .await
        .context("Failed to exchange Auth0 authorization code for tokens")?;

    if !response.status().is_success() {
        let status = response.status();
        let body = response.text().await.unwrap_or_default();
        anyhow::bail!("Auth0 PKCE token exchange failed ({}): {}", status, body);
    }

    let token_json: serde_json::Value = response
        .json()
        .await
        .context("Failed to parse Auth0 PKCE token response")?;

    if let Some(id_token) = token_json.get("id_token").and_then(|v| v.as_str()) {
        return Ok(id_token.to_string());
    }
    if let Some(access_token) = token_json.get("access_token").and_then(|v| v.as_str()) {
        return Ok(access_token.to_string());
    }
    anyhow::bail!("Auth0 PKCE response missing id_token/access_token.");
}

async fn auth0_device_authorize(
    domain: &str,
    client_id: &str,
    audience: Option<&str>,
    open: bool,
    json: bool,
) -> Result<String> {
    let http_client = Client::new();
    let device_url = format!("https://{}/oauth/device/code", domain.trim());
    let token_url = format!("https://{}/oauth/token", domain.trim());

    let mut form: Vec<(String, String)> = vec![
        ("client_id".to_string(), client_id.to_string()),
        ("scope".to_string(), "openid profile email".to_string()),
    ];
    if let Some(aud) = audience {
        if !aud.trim().is_empty() {
            form.push(("audience".to_string(), aud.trim().to_string()));
        }
    }

    let response = http_client
        .post(&device_url)
        .form(&form)
        .send()
        .await
        .context("Failed to request Auth0 device code")?;

    if !response.status().is_success() {
        let status = response.status();
        let body = response.text().await.unwrap_or_default();
        anyhow::bail!("Auth0 device code request failed ({}): {}", status, body);
    }

    let device_json: serde_json::Value = response
        .json()
        .await
        .context("Failed to parse Auth0 device code response")?;

    let device_code = device_json
        .get("device_code")
        .and_then(|v| v.as_str())
        .ok_or_else(|| anyhow::anyhow!("Auth0 device response missing device_code"))?
        .to_string();

    let verification_uri_complete = device_json
        .get("verification_uri_complete")
        .and_then(|v| v.as_str())
        .map(|s| s.to_string());
    let verification_uri = device_json
        .get("verification_uri")
        .and_then(|v| v.as_str())
        .map(|s| s.to_string());
    let user_code = device_json
        .get("user_code")
        .and_then(|v| v.as_str())
        .map(|s| s.to_string());
    let expires_in = device_json.get("expires_in").and_then(|v| v.as_u64()).unwrap_or(600);
    let mut interval = device_json.get("interval").and_then(|v| v.as_u64()).unwrap_or(5).max(1);

    let login_url = verification_uri_complete
        .or(verification_uri)
        .ok_or_else(|| anyhow::anyhow!("Auth0 device response missing verification URL"))?;

    if !json {
        println!("Open this URL to authenticate:");
        println!("  {}", login_url);
        if let Some(code) = user_code.as_deref() {
            println!("User code: {}", code);
        }
        if open {
            match open_browser(&login_url) {
                Ok(()) => println!("Opened browser for authentication."),
                Err(err) => eprintln!("Could not open browser automatically: {}", err),
            }
        }
        println!("Waiting for authentication confirmation...");
    }

    let started = Instant::now();
    loop {
        if started.elapsed() > Duration::from_secs(expires_in) {
            anyhow::bail!("Auth0 device login timed out (device code expired).");
        }

        tokio::time::sleep(Duration::from_secs(interval)).await;

        let response = http_client
            .post(&token_url)
            .form(&[
                ("grant_type", "urn:ietf:params:oauth:grant-type:device_code"),
                ("device_code", device_code.as_str()),
                ("client_id", client_id),
            ])
            .send()
            .await
            .context("Failed to poll Auth0 device token endpoint")?;

        if response.status().is_success() {
            let token_json: serde_json::Value =
                response.json().await.context("Failed to parse Auth0 token response")?;
            if let Some(id_token) = token_json.get("id_token").and_then(|v| v.as_str()) {
                return Ok(id_token.to_string());
            }
            if let Some(access_token) = token_json.get("access_token").and_then(|v| v.as_str()) {
                return Ok(access_token.to_string());
            }
            anyhow::bail!("Auth0 token response missing id_token/access_token.");
        }

        let status = response.status();
        let body: serde_json::Value = response.json().await.unwrap_or_default();
        let err = body.get("error").and_then(|v| v.as_str()).unwrap_or_default();
        match err {
            "authorization_pending" => continue,
            "slow_down" => {
                interval += 5;
                continue;
            }
            "expired_token" => anyhow::bail!("Auth0 device code expired. Run login again."),
            "access_denied" => anyhow::bail!("Auth0 login denied by user."),
            _ => {
                let desc = body
                    .get("error_description")
                    .and_then(|v| v.as_str())
                    .unwrap_or("unknown error");
                anyhow::bail!("Auth0 device login failed ({}): {} {}", status, err, desc);
            }
        }
    }
}

async fn handle_auth_command(config: &mut Config, base_url: &str, command: AuthCommand) -> Result<()> {
    match command {
        AuthCommand::Login {
            auth0,
            auth0_token,
            auth0_callback_url,
            auth0_domain,
            auth0_redirect_uri,
            auth0_audience,
            auth0_client_secret,
            browser: use_browser_flow,
            browser_pkce: use_pkce_flow,
            pkce_port,
            no_browser_open,
            password: use_password_grant,
            session_duration_hours,
            context,
            no_save,
            json,
        } => {
            let mut token =
                auth0_token.or_else(|| auth0_callback_url.as_deref().and_then(extract_token_from_callback_url));

            if token.is_none() && use_pkce_flow {
                if pkce_port == 0 {
                    anyhow::bail!("--pkce-port cannot be 0.");
                }
                let ctx_name_ref = context.as_deref().or_else(|| config.current_context.as_deref());
                let ctx_ref = ctx_name_ref.and_then(|n| config.contexts.get(n));
                let domain = auth0_domain
                    .as_deref()
                    .or_else(|| ctx_ref.and_then(|c| c.gateway_auth0_domain.as_deref()))
                    .ok_or_else(|| {
                        anyhow::anyhow!("Auth0 domain not set. Use 'noetl context add --auth0-domain <domain>' first.")
                    })?;
                let client_id = ctx_ref
                    .and_then(|c| c.gateway_auth0_client_id.as_deref())
                    .ok_or_else(|| {
                        anyhow::anyhow!(
                            "Auth0 client_id not set. Use 'noetl context add --auth0-client-id <id>' first."
                        )
                    })?;
                let context_redirect_uri = ctx_ref.and_then(|c| c.gateway_auth0_redirect_uri.as_deref());
                let redirect_uri =
                    resolve_pkce_redirect_uri(auth0_redirect_uri.as_deref(), context_redirect_uri, pkce_port)?;
                let audience = auth0_audience
                    .as_deref()
                    .or_else(|| ctx_ref.and_then(|c| c.gateway_auth0_audience.as_deref()));
                let env_client_secret = std::env::var("NOETL_AUTH0_CLIENT_SECRET")
                    .ok()
                    .filter(|s| !s.trim().is_empty());
                let client_secret = auth0_client_secret
                    .as_deref()
                    .or_else(|| env_client_secret.as_deref())
                    .or_else(|| ctx_ref.and_then(|c| c.gateway_auth0_client_secret.as_deref()))
                    .filter(|s| !s.trim().is_empty());
                let login_hint = auth0.as_deref().filter(|v| looks_like_email(v)).map(str::trim);

                token = Some(
                    auth0_pkce_authorize(
                        domain,
                        client_id,
                        &redirect_uri,
                        audience,
                        client_secret,
                        login_hint,
                        !no_browser_open,
                        json,
                    )
                    .await
                    .context("Auth0 browser PKCE login failed")?,
                );
            }

            if token.is_none() && use_browser_flow {
                let ctx_name_ref = context.as_deref().or_else(|| config.current_context.as_deref());
                let ctx_ref = ctx_name_ref.and_then(|n| config.contexts.get(n));
                let domain = auth0_domain
                    .as_deref()
                    .or_else(|| ctx_ref.and_then(|c| c.gateway_auth0_domain.as_deref()))
                    .ok_or_else(|| {
                        anyhow::anyhow!("Auth0 domain not set. Use 'noetl context add --auth0-domain <domain>' first.")
                    })?;
                let client_id = ctx_ref
                    .and_then(|c| c.gateway_auth0_client_id.as_deref())
                    .ok_or_else(|| {
                        anyhow::anyhow!(
                            "Auth0 client_id not set. Use 'noetl context add --auth0-client-id <id>' first."
                        )
                    })?;
                let audience = auth0_audience
                    .as_deref()
                    .or_else(|| ctx_ref.and_then(|c| c.gateway_auth0_audience.as_deref()));
                token = Some(
                    auth0_device_authorize(domain, client_id, audience, !no_browser_open, json)
                        .await
                        .context("Auth0 browser/device login failed")?,
                );
            }

            if token.is_none() {
                if let Some(auth0_input) = auth0 {
                    if let Some(parsed) = extract_token_from_callback_url(&auth0_input) {
                        token = Some(parsed);
                    } else if looks_like_jwt(&auth0_input) {
                        token = Some(auth0_input.trim().to_string());
                    } else if looks_like_email(&auth0_input) {
                        // Resolve context config
                        let ctx_name_ref = context.as_deref().or_else(|| config.current_context.as_deref());
                        let ctx_ref = ctx_name_ref.and_then(|n| config.contexts.get(n));
                        let early_domain = auth0_domain
                            .as_deref()
                            .or_else(|| ctx_ref.and_then(|c| c.gateway_auth0_domain.as_deref()));
                        let early_audience = auth0_audience
                            .as_deref()
                            .or_else(|| ctx_ref.and_then(|c| c.gateway_auth0_audience.as_deref()));
                        let env_client_secret = std::env::var("NOETL_AUTH0_CLIENT_SECRET")
                            .ok()
                            .filter(|s| !s.trim().is_empty());
                        let early_client_id = ctx_ref.and_then(|c| c.gateway_auth0_client_id.as_deref());
                        let early_client_secret = auth0_client_secret
                            .as_deref()
                            .or_else(|| env_client_secret.as_deref())
                            .or_else(|| ctx_ref.and_then(|c| c.gateway_auth0_client_secret.as_deref()))
                            .filter(|s| !s.trim().is_empty());
                        let early_redirect_uri = ctx_ref.and_then(|c| c.gateway_auth0_redirect_uri.as_deref());

                        if use_password_grant || early_client_secret.is_some() {
                            // Password grant: prompt for password, exchange with Auth0 directly
                            let domain = early_domain.ok_or_else(|| {
                                anyhow::anyhow!(
                                    "Auth0 domain not set. Use 'noetl context add --auth0-domain <domain>' first."
                                )
                            })?;
                            let client_id = early_client_id.ok_or_else(|| {
                                anyhow::anyhow!(
                                    "Auth0 client_id not set. Use 'noetl context add --auth0-client-id <id>' first."
                                )
                            })?;
                            if !json {
                                eprint!("Password for {}: ", auth0_input.trim());
                            }
                            let pwd = rpassword::read_password().context("Failed to read password")?;
                            if pwd.is_empty() {
                                anyhow::bail!("Password cannot be empty.");
                            }
                            let token_endpoint = format!("https://{}/oauth/token", domain);
                            if !json {
                                println!("Authenticating with Auth0...");
                            }
                            let http_client = Client::new();
                            let mut auth_payload = serde_json::json!({
                                "grant_type": "password",
                                "client_id": client_id,
                                "username": auth0_input.trim(),
                                "password": pwd,
                                "scope": "openid profile email"
                            });
                            if let Some(secret) = early_client_secret {
                                if !secret.trim().is_empty() {
                                    auth_payload["client_secret"] =
                                        serde_json::Value::String(secret.trim().to_string());
                                }
                            }
                            if let Some(aud) = early_audience {
                                if !aud.trim().is_empty() {
                                    auth_payload["audience"] = serde_json::Value::String(aud.trim().to_string());
                                }
                            }
                            let resp = http_client
                                .post(&token_endpoint)
                                .json(&auth_payload)
                                .send()
                                .await
                                .context("Failed to reach Auth0 token endpoint")?;
                            if !resp.status().is_success() {
                                let status = resp.status();
                                let body: serde_json::Value = resp.json().await.unwrap_or_default();
                                let desc = body
                                    .get("error_description")
                                    .and_then(|v| v.as_str())
                                    .unwrap_or("unknown error");
                                anyhow::bail!("Auth0 password grant failed ({}): {}", status, desc);
                            }
                            let auth0_resp: serde_json::Value =
                                resp.json().await.context("Failed to parse Auth0 token response")?;
                            let id_token = auth0_resp.get("id_token")
                                .and_then(|v| v.as_str())
                                .ok_or_else(|| anyhow::anyhow!(
                                    "Auth0 response missing id_token. The application may not be authorized for password + openid scope."
                                ))?;
                            token = Some(id_token.to_string());
                        } else {
                            // Browser flow: print authorization URL
                            let early_redirect_uri = early_redirect_uri;
                            if !json {
                                println!("Auth0 login hint: {}", auth0_input.trim());
                                if let (Some(domain), Some(client_id)) = (early_domain, early_client_id) {
                                    let nonce = std::time::SystemTime::now()
                                        .duration_since(std::time::UNIX_EPOCH)
                                        .unwrap_or_default()
                                        .as_millis()
                                        .to_string();
                                    let redirect = early_redirect_uri.unwrap_or("https://your-app/login");
                                    let encoded_redirect = redirect.replace(':', "%3A").replace('/', "%2F");
                                    let encoded_hint = auth0_input.trim().replace('@', "%40");
                                    println!("Open in browser to authenticate:");
                                    println!(
                                        "  https://{}/authorize?response_type=id_token%20token&client_id={}&redirect_uri={}&scope=openid%20profile%20email&nonce={}&login_hint={}",
                                        domain, client_id, encoded_redirect, nonce, encoded_hint
                                    );
                                    println!("After login, copy the full redirect URL (contains #id_token=...) and paste below.");
                                } else {
                                    println!("Open your browser and log in via the gateway or Auth0 UI.");
                                    println!(
                                        "After login, copy the callback URL (contains #id_token=...) and paste below."
                                    );
                                }
                                println!("Paste callback URL or token:");
                            }
                            let provided = prompt_input("> ")?;
                            token = extract_token_from_callback_url(&provided)
                                .or_else(|| (!provided.is_empty()).then_some(provided));
                        }
                    } else if !auth0_input.trim().is_empty() {
                        token = Some(auth0_input.trim().to_string());
                    }
                }
            }

            let token = token.ok_or_else(|| {
                anyhow::anyhow!(
                    "Missing Auth0 token. Use --auth0-token, --auth0-callback-url, --browser, --browser-pkce, or --auth0 <email|token|callback-url>."
                )
            })?;

            let target_context_name = context.or_else(|| config.current_context.clone());
            let context_auth0_domain = target_context_name
                .as_ref()
                .and_then(|name| config.contexts.get(name))
                .and_then(|ctx| ctx.gateway_auth0_domain.clone());
            let effective_auth0_domain = auth0_domain.or(context_auth0_domain);

            let gateway_base = base_url.trim_end_matches('/');
            let login_url = format!("{}/api/auth/login", gateway_base);
            let client = Client::new();

            let mut payload = serde_json::json!({
                "auth0_token": token,
                "session_duration_hours": session_duration_hours
            });
            if let Some(domain) = effective_auth0_domain.clone() {
                payload["auth0_domain"] = serde_json::Value::String(domain);
            }

            let response = client
                .post(&login_url)
                .json(&payload)
                .send()
                .await
                .context("Failed to call gateway auth login endpoint")?;

            if !response.status().is_success() {
                let status = response.status();
                let text = response.text().await.unwrap_or_default();
                anyhow::bail!(
                    "Auth login failed at {}: {} - {}. Ensure this server URL points to Gateway (not NoETL API) and token/domain are valid.",
                    login_url,
                    status,
                    text
                );
            }

            let login_result: serde_json::Value = response
                .json()
                .await
                .context("Failed to parse gateway auth login response")?;

            let session_token = login_result
                .get("session_token")
                .and_then(|v| v.as_str())
                .ok_or_else(|| anyhow::anyhow!("Gateway response missing session_token"))?
                .to_string();

            let mut saved_context = None::<String>;
            if !no_save {
                if let Some(context_name) = target_context_name.clone() {
                    if let Some(ctx) = config.contexts.get_mut(&context_name) {
                        ctx.gateway_session_token = Some(session_token.clone());
                        if effective_auth0_domain.is_some() {
                            ctx.gateway_auth0_domain = effective_auth0_domain.clone();
                        }
                        config.save()?;
                        saved_context = Some(context_name);
                    } else {
                        eprintln!(
                            "Warning: context '{}' not found, token not saved to config.",
                            context_name
                        );
                    }
                } else {
                    eprintln!("Warning: no current context selected, token not saved to config.");
                }
            }

            if json {
                let output = serde_json::json!({
                    "gateway_url": gateway_base,
                    "saved_context": saved_context,
                    "session_token": session_token,
                    "login": login_result,
                });
                println!("{}", serde_json::to_string(&output)?);
            } else {
                let user_email = login_result
                    .get("user")
                    .and_then(|u| u.get("email"))
                    .and_then(|v| v.as_str())
                    .unwrap_or("unknown");
                println!("Gateway login successful.");
                println!("  Gateway: {}", gateway_base);
                println!("  User:    {}", user_email);
                println!("  Token:   {}", mask_token(&session_token));
                if let Some(context_name) = saved_context {
                    println!("  Saved:   context '{}'", context_name);
                } else if no_save {
                    println!("  Saved:   disabled (--no-save)");
                } else {
                    println!("  Saved:   no");
                }
                println!("\nUse this session for CLI calls:");
                println!("  export NOETL_SESSION_TOKEN='{}'", session_token);
            }
        }
        AuthCommand::Logout { context } => {
            let target_context_name = context.or_else(|| config.current_context.clone());
            let Some(context_name) = target_context_name else {
                anyhow::bail!("No context selected. Use --context <name> or 'noetl context use <name>'.");
            };

            let Some(ctx) = config.contexts.get_mut(&context_name) else {
                anyhow::bail!("Context '{}' not found.", context_name);
            };

            ctx.gateway_session_token = None;
            config.save()?;
            println!("Cleared cached gateway session token for context '{}'.", context_name);
        }
    }
    Ok(())
}

async fn register_directory(
    client: &Client,
    base_url: &str,
    use_gateway_proxy: bool,
    resource_type: &str,
    directory: &PathBuf,
    extensions: &[&str],
) -> Result<()> {
    let files = scan_directory(directory, extensions)?;

    if files.is_empty() {
        println!("No {} files found in directory: {:?}", extensions.join(", "), directory);
        return Ok(());
    }

    println!("Found {} file(s) in {:?}", files.len(), directory);

    let mut success_count = 0;
    let mut fail_count = 0;

    for file in files {
        match register_resource(client, base_url, use_gateway_proxy, resource_type, &file).await {
            Ok(_) => success_count += 1,
            Err(e) => {
                eprintln!("Failed to register {:?}: {}", file, e);
                fail_count += 1;
            }
        }
    }

    println!(
        "\nRegistration complete: {} succeeded, {} failed",
        success_count, fail_count
    );
    Ok(())
}

fn scan_directory(directory: &PathBuf, extensions: &[&str]) -> Result<Vec<PathBuf>> {
    let mut files = Vec::new();

    if !directory.exists() {
        return Err(anyhow::anyhow!("Directory does not exist: {:?}", directory));
    }

    if !directory.is_dir() {
        return Err(anyhow::anyhow!("Path is not a directory: {:?}", directory));
    }

    scan_directory_recursive(directory, extensions, &mut files)?;
    Ok(files)
}

fn scan_directory_recursive(dir: &PathBuf, extensions: &[&str], files: &mut Vec<PathBuf>) -> Result<()> {
    for entry in fs::read_dir(dir).context(format!("Failed to read directory: {:?}", dir))? {
        let entry = entry?;
        let path = entry.path();

        if path.is_dir() {
            scan_directory_recursive(&path, extensions, files)?;
        } else if path.is_file() {
            if let Some(ext) = path.extension() {
                if let Some(ext_str) = ext.to_str() {
                    if extensions.contains(&ext_str) {
                        // Validate file content before adding
                        if is_valid_resource_file(&path, extensions)? {
                            files.push(path);
                        }
                    }
                }
            }
        }
    }
    Ok(())
}

fn is_valid_resource_file(file: &PathBuf, extensions: &[&str]) -> Result<bool> {
    let content = fs::read_to_string(file).context(format!("Failed to read file: {:?}", file))?;

    // Check if it's a YAML file (playbook)
    if extensions.contains(&"yaml") || extensions.contains(&"yml") {
        // Look for apiVersion and kind: Playbook
        if content.contains("apiVersion:") && content.contains("kind: Playbook") {
            return Ok(true);
        }
        return Ok(false);
    }

    // Check if it's a JSON file (credential)
    if extensions.contains(&"json") {
        // Try to parse as JSON and check for "type" field
        match serde_json::from_str::<serde_json::Value>(&content) {
            Ok(json) => {
                if json.get("type").is_some() {
                    return Ok(true);
                }
            }
            Err(_) => return Ok(false),
        }
    }

    Ok(false)
}

async fn register_resource(
    client: &Client,
    base_url: &str,
    use_gateway_proxy: bool,
    resource_type: &str,
    file: &PathBuf,
) -> Result<()> {
    let content = fs::read_to_string(file).context(format!("Failed to read file: {:?}", file))?;

    let (url, request_body) = if resource_type == "Credential" {
        // For credentials, parse JSON and POST to /api/credentials
        let credential_data: serde_json::Value =
            serde_json::from_str(&content).context(format!("Failed to parse credential JSON from file: {:?}", file))?;

        (api_url(base_url, "credentials", use_gateway_proxy), credential_data)
    } else {
        // For playbooks, base64 encode and POST to /api/catalog/register
        let content_base64 = BASE64_STANDARD.encode(&content);
        let request = RegisterRequest {
            content: content_base64,
            resource_type: resource_type.to_string(),
        };

        (
            api_url(base_url, "catalog/register", use_gateway_proxy),
            serde_json::to_value(request)?,
        )
    };

    let response = client
        .post(&url)
        .json(&request_body)
        .send()
        .await
        .context("Failed to send register request")?;

    if response.status().is_success() {
        let result: serde_json::Value = response.json().await?;
        println!("{} registered successfully: {}", resource_type, result);
    } else {
        let status = response.status();
        // 401 against a gateway-proxied register call almost always
        // means the cached session token expired — point the
        // operator at the right re-auth command before falling
        // through to the generic error path.
        check_gateway_auth_expired(status, use_gateway_proxy, None);
        let text = response.text().await?;
        eprintln!("Failed to register {}: {} - {}", resource_type, status, text);
        std::process::exit(1);
    }

    Ok(())
}

/// Execute playbook on distributed server-worker environment
async fn execute_playbook_distributed(
    client: &Client,
    base_url: &str,
    use_gateway_proxy: bool,
    path: &str,
    version: Option<i64>,
    workload: serde_json::Value,
    resource_kind: ExecutionResourceKind,
    json_only: bool,
) -> Result<()> {
    let url = api_url(base_url, "execute", use_gateway_proxy);
    let request_body = ExecuteApiRequest {
        path: path.to_string(),
        version,
        workload,
        resource_kind,
    };

    if !json_only {
        println!("Executing playbook on distributed server...");
        println!("  Path: {}", path);
        if let Some(v) = version {
            println!("  Version: {}", v);
        }
        println!("  Server: {}", base_url);
    }

    let response = client
        .post(&url)
        .json(&request_body)
        .send()
        .await
        .context("Failed to send execute request")?;

    if response.status().is_success() {
        let result: serde_json::Value = response.json().await?;
        if json_only {
            println!("{}", serde_json::to_string(&result)?);
        } else {
            println!("\nExecution started:");
            println!("{}", serde_json::to_string_pretty(&result)?);

            // Extract execution_id if available and show status command hint
            if let Some(exec_id) = result.get("execution_id") {
                println!("\nTo check status:");
                println!("  noetl execute status {}", exec_id);
            }
        }
    } else {
        let status = response.status();
        check_gateway_auth_expired(status, use_gateway_proxy, None);
        let text = response.text().await?;
        eprintln!("Failed to execute playbook: {} - {}", status, text);
        std::process::exit(1);
    }

    Ok(())
}

async fn execute_rerun_distributed(
    client: &Client,
    base_url: &str,
    use_gateway_proxy: bool,
    execution_id: &str,
    workload: serde_json::Value,
    resource_kind: ExecutionResourceKind,
    json_only: bool,
) -> Result<()> {
    let url = api_url(
        base_url,
        &format!("executions/{}/rerun", execution_id),
        use_gateway_proxy,
    );
    let request_body = RerunApiRequest {
        workload,
        resource_kind,
    };

    if !json_only {
        println!("Rerunning execution on distributed server...");
        println!("  Source Execution: {}", execution_id);
        println!("  Server: {}", base_url);
    }

    let response = client
        .post(&url)
        .json(&request_body)
        .send()
        .await
        .context("Failed to send rerun request")?;

    if response.status().is_success() {
        let result: serde_json::Value = response.json().await?;
        if json_only {
            println!("{}", serde_json::to_string(&result)?);
        } else {
            println!("\nRerun started:");
            println!("{}", serde_json::to_string_pretty(&result)?);
            if let Some(exec_id) = result.get("execution_id") {
                println!("\nTo check status:");
                println!("  noetl execute status {}", exec_id);
            }
        }
    } else {
        let status = response.status();
        let text = response.text().await?;
        eprintln!("Failed to rerun execution: {} - {}", status, text);
        std::process::exit(1);
    }

    Ok(())
}

async fn execute_playbook(
    client: &Client,
    base_url: &str,
    use_gateway_proxy: bool,
    path: &str,
    input: Option<PathBuf>,
    json_only: bool,
) -> Result<()> {
    // Legacy function - delegate to new one without version
    let payload = build_distributed_payload(input.as_ref(), None, &[])?;
    execute_playbook_distributed(
        client,
        base_url,
        use_gateway_proxy,
        path,
        None,
        payload,
        ExecutionResourceKind::Playbook,
        json_only,
    )
    .await
}

async fn get_status(
    client: &Client,
    base_url: &str,
    use_gateway_proxy: bool,
    execution_id: &str,
    json_only: bool,
) -> Result<()> {
    fn parse_status_timestamp(value: &str) -> Option<chrono::DateTime<chrono::Utc>> {
        if let Ok(parsed) = chrono::DateTime::parse_from_rfc3339(value) {
            return Some(parsed.with_timezone(&chrono::Utc));
        }
        if let Ok(naive) = chrono::NaiveDateTime::parse_from_str(value, "%Y-%m-%dT%H:%M:%S%.f") {
            return Some(chrono::DateTime::<chrono::Utc>::from_naive_utc_and_offset(
                naive,
                chrono::Utc,
            ));
        }
        None
    }

    fn format_duration_human(total_seconds: f64) -> String {
        let seconds = total_seconds.max(0.0).round() as i64;
        let days = seconds / 86_400;
        let hours = (seconds % 86_400) / 3_600;
        let minutes = (seconds % 3_600) / 60;
        let secs = seconds % 60;

        let mut parts: Vec<String> = Vec::new();
        if days > 0 {
            parts.push(format!("{}d", days));
        }
        if hours > 0 {
            parts.push(format!("{}h", hours));
        }
        if minutes > 0 {
            parts.push(format!("{}m", minutes));
        }
        if secs > 0 || parts.is_empty() {
            parts.push(format!("{}s", secs));
        }
        parts.join(" ")
    }

    let infer_duration_human = |result: &serde_json::Value| -> Option<String> {
        if let Some(human) = result.get("duration_human").and_then(|v| v.as_str()) {
            if !human.is_empty() {
                return Some(human.to_string());
            }
        }

        let seconds = result.get("duration_seconds").and_then(|v| v.as_f64()).or_else(|| {
            result
                .get("duration_seconds")
                .and_then(|v| v.as_i64())
                .map(|v| v as f64)
        });
        if let Some(total_seconds) = seconds {
            return Some(format_duration_human(total_seconds));
        }

        let start = result
            .get("start_time")
            .and_then(|v| v.as_str())
            .and_then(parse_status_timestamp);
        let end = result
            .get("end_time")
            .and_then(|v| v.as_str())
            .and_then(parse_status_timestamp);
        let completed = result.get("completed").and_then(|v| v.as_bool()).unwrap_or(false);

        if let Some(start_dt) = start {
            let effective_end = end.or_else(|| if completed { None } else { Some(chrono::Utc::now()) });
            if let Some(end_dt) = effective_end {
                let total_seconds = (end_dt - start_dt).num_milliseconds().max(0) as f64 / 1000.0;
                return Some(format_duration_human(total_seconds));
            }
        }

        None
    };

    let summarize_and_print = |result: &serde_json::Value, show_fallback_note: bool| {
        // Show a concise summary by default
        let completed = result.get("completed").and_then(|v| v.as_bool()).unwrap_or(false);
        let failed = result.get("failed").and_then(|v| v.as_bool()).unwrap_or(false);
        let current_step = result.get("current_step").and_then(|v| v.as_str()).unwrap_or("unknown");
        let completed_steps = result
            .get("completed_steps")
            .and_then(|v| v.as_array())
            .map(|a| a.len())
            .unwrap_or(0);

        let status_str = if completed {
            if failed {
                "FAILED"
            } else {
                "COMPLETED"
            }
        } else {
            "RUNNING"
        };

        let status_color = if completed {
            if failed {
                "\x1b[31m"
            } else {
                "\x1b[32m"
            } // red or green
        } else {
            "\x1b[33m" // yellow
        };

        println!("\n{}{}\x1b[0m", status_color, "=".repeat(60));
        println!("Execution: {}", execution_id);
        println!("Status:    {}{}\x1b[0m", status_color, status_str);
        println!("Steps:     {} completed", completed_steps);
        if let Some(duration_human) = infer_duration_human(result) {
            println!("Duration:  {}", duration_human);
        }
        if !completed {
            println!("Current:   {}", current_step);
        }

        if failed {
            if let Some(error) = result.get("error").and_then(|v| v.as_str()) {
                println!("\x1b[31mError:\x1b[0m     {}", error);
            }
        }

        if let Some(steps) = result.get("completed_steps").and_then(|v| v.as_array()) {
            if !steps.is_empty() {
                println!("\nCompleted steps:");
                for step in steps {
                    if let Some(step_name) = step.as_str() {
                        println!("  - {}", step_name);
                    }
                }
            }
        }

        println!("{}{}\x1b[0m\n", status_color, "=".repeat(60));
        if show_fallback_note {
            println!("Note: status served from execution event log fallback (server state cache miss).");
        }
        println!("Use --json for full execution details");
    };

    let status_url = api_url(
        base_url,
        &format!("executions/{}/status", execution_id),
        use_gateway_proxy,
    );
    let response = client
        .get(&status_url)
        .send()
        .await
        .context("Failed to send status request")?;

    if response.status().is_success() {
        let result: serde_json::Value = response.json().await?;
        if json_only {
            println!("{}", serde_json::to_string(&result)?);
        } else {
            summarize_and_print(&result, false);
        }
        return Ok(());
    }

    if response.status() == reqwest::StatusCode::NOT_FOUND {
        let fallback_url = api_url(
            base_url,
            &format!("executions/{}?page=1&page_size=10", execution_id),
            use_gateway_proxy,
        );
        let fallback_response = client
            .get(&fallback_url)
            .send()
            .await
            .context("Failed to send fallback execution request")?;

        if fallback_response.status().is_success() {
            let execution_doc: serde_json::Value = fallback_response.json().await?;
            let fallback_status = execution_doc
                .get("status")
                .and_then(|v| v.as_str())
                .unwrap_or("UNKNOWN");
            let latest_node = execution_doc
                .get("events")
                .and_then(|v| v.as_array())
                .and_then(|events| events.first())
                .and_then(|event| event.get("node_name"))
                .and_then(|v| v.as_str())
                .unwrap_or("unknown");
            let completed = matches!(fallback_status, "COMPLETED" | "FAILED" | "CANCELLED");
            let failed = fallback_status == "FAILED";
            let start_time = execution_doc
                .get("start_time")
                .cloned()
                .unwrap_or(serde_json::Value::Null);
            let end_time = execution_doc
                .get("end_time")
                .cloned()
                .unwrap_or(serde_json::Value::Null);
            let duration_seconds = execution_doc
                .get("duration_seconds")
                .cloned()
                .unwrap_or(serde_json::Value::Null);
            let duration_human = execution_doc
                .get("duration_human")
                .cloned()
                .unwrap_or(serde_json::Value::Null);

            let normalized = serde_json::json!({
                "execution_id": execution_id,
                "current_step": latest_node,
                "completed_steps": [],
                "failed": failed,
                "completed": completed,
                "completion_inferred": true,
                "variables": {},
                "status": fallback_status,
                "source": "executions_endpoint_fallback",
                "start_time": start_time,
                "end_time": end_time,
                "duration_seconds": duration_seconds,
                "duration_human": duration_human,
            });

            if json_only {
                println!("{}", serde_json::to_string(&normalized)?);
            } else {
                summarize_and_print(&normalized, true);
            }
            return Ok(());
        }

        let fallback_status = fallback_response.status();
        let fallback_text = fallback_response.text().await?;
        eprintln!(
            "Failed to get status: {} - {}; fallback failed: {} - {}",
            response.status(),
            "Execution not found in server state cache",
            fallback_status,
            fallback_text
        );
        return Ok(());
    }

    let status = response.status();
    let text = response.text().await?;
    eprintln!("Failed to get status: {} - {}", status, text);
    Ok(())
}

async fn cancel_execution(
    client: &Client,
    base_url: &str,
    use_gateway_proxy: bool,
    execution_id: &str,
    reason: Option<String>,
    cascade: bool,
    json_only: bool,
) -> Result<()> {
    let url = api_url(
        base_url,
        &format!("executions/{}/cancel", execution_id),
        use_gateway_proxy,
    );

    // Build request body
    let mut body = serde_json::json!({
        "cascade": cascade
    });
    if let Some(r) = &reason {
        body["reason"] = serde_json::json!(r);
    }

    let response = client
        .post(&url)
        .json(&body)
        .send()
        .await
        .context("Failed to send cancel request")?;

    if response.status().is_success() {
        let result: serde_json::Value = response.json().await?;
        if json_only {
            println!("{}", serde_json::to_string(&result)?);
        } else {
            let status = result.get("status").and_then(|v| v.as_str()).unwrap_or("unknown");
            let cancelled_count = result
                .get("cancelled_executions")
                .and_then(|v| v.as_array())
                .map(|a| a.len())
                .unwrap_or(0);
            let message = result.get("message").and_then(|v| v.as_str()).unwrap_or("");

            println!("\n\x1b[33m{}\x1b[0m", "=".repeat(60));
            println!("Execution: {}", execution_id);
            println!("Status:    \x1b[33m{}\x1b[0m", status.to_uppercase());
            println!("Cancelled: {} execution(s)", cancelled_count);
            if !message.is_empty() {
                println!("Message:   {}", message);
            }
            if let Some(r) = &reason {
                println!("Reason:    {}", r);
            }

            // List cancelled executions
            if let Some(cancelled) = result.get("cancelled_executions").and_then(|v| v.as_array()) {
                if cancelled.len() > 1 {
                    println!("\nCancelled executions:");
                    for exec_id in cancelled {
                        if let Some(id) = exec_id.as_str() {
                            println!("  - {}", id);
                        }
                    }
                }
            }

            println!("\x1b[33m{}\x1b[0m\n", "=".repeat(60));
        }
    } else if response.status().as_u16() == 400 {
        // Already completed/cancelled
        let result: serde_json::Value = response.json().await?;
        if json_only {
            println!("{}", serde_json::to_string(&result)?);
        } else {
            let message = result
                .get("message")
                .and_then(|v| v.as_str())
                .unwrap_or("Cannot cancel execution");
            println!("\x1b[33mWarning:\x1b[0m {}", message);
        }
    } else {
        let status = response.status();
        let text = response.text().await?;
        eprintln!("Failed to cancel execution: {} - {}", status, text);
        std::process::exit(1);
    }
    Ok(())
}

async fn cleanup_stuck_executions(
    client: &Client,
    base_url: &str,
    use_gateway_proxy: bool,
    older_than_minutes: u32,
    dry_run: bool,
    json_only: bool,
) -> Result<()> {
    let url = api_url(base_url, "executions/cleanup", use_gateway_proxy);

    // Build request body
    let body = serde_json::json!({
        "older_than_minutes": older_than_minutes,
        "dry_run": dry_run
    });

    let response = client
        .post(&url)
        .json(&body)
        .send()
        .await
        .context("Failed to send cleanup request")?;

    if response.status().is_success() {
        let result: serde_json::Value = response.json().await?;

        if json_only {
            println!("{}", serde_json::to_string(&result)?);
        } else {
            let cancelled_count = result.get("cancelled_count").and_then(|v| v.as_u64()).unwrap_or(0);
            let message = result.get("message").and_then(|v| v.as_str()).unwrap_or("");
            let execution_ids = result
                .get("execution_ids")
                .and_then(|v| v.as_array())
                .map(|arr| arr.iter().filter_map(|v| v.as_i64()).collect::<Vec<_>>())
                .unwrap_or_default();

            println!("\n\x1b[36m{}\x1b[0m", "=".repeat(60));
            if dry_run {
                println!("\x1b[36m[DRY RUN]\x1b[0m Cleanup Stuck Executions");
            } else {
                println!("Cleanup Stuck Executions");
            }
            println!("\x1b[36m{}\x1b[0m", "=".repeat(60));
            println!("Older than:    {} minutes", older_than_minutes);
            println!("Found:         {} stuck execution(s)", cancelled_count);

            if !message.is_empty() {
                println!("\n{}", message);
            }

            if !execution_ids.is_empty() {
                println!("\nExecution IDs:");
                for exec_id in execution_ids {
                    println!("  - {}", exec_id);
                }
            }

            if dry_run && cancelled_count > 0 {
                println!("\n\x1b[33mNote:\x1b[0m Run without --dry-run to actually cancel these executions");
            }

            println!("\x1b[36m{}\x1b[0m\n", "=".repeat(60));
        }
    } else {
        let status = response.status();
        let text = response.text().await?;
        eprintln!("Failed to cleanup stuck executions: {} - {}", status, text);
        std::process::exit(1);
    }

    Ok(())
}

async fn list_resources(
    client: &Client,
    base_url: &str,
    use_gateway_proxy: bool,
    resource_type: &str,
    json_only: bool,
) -> Result<()> {
    let url = api_url(base_url, "catalog/list", use_gateway_proxy);
    let body = serde_json::json!({"resource_type": resource_type});
    let response = client
        .post(&url)
        .json(&body)
        .send()
        .await
        .context("Failed to send list request")?;

    if response.status().is_success() {
        let result: serde_json::Value = response.json().await?;
        if json_only {
            println!("{}", serde_json::to_string(&result)?);
        } else {
            println!("Catalog ({}):", resource_type);
            println!("{}", serde_json::to_string_pretty(&result)?);
        }
    } else {
        let status = response.status();
        check_gateway_auth_expired(status, use_gateway_proxy, None);
        let text = response.text().await?;
        eprintln!("Failed to list resources: {} - {}", status, text);
    }
    Ok(())
}

async fn get_catalog_resource(client: &Client, base_url: &str, use_gateway_proxy: bool, path: &str) -> Result<()> {
    let url = api_url(base_url, "catalog/resource", use_gateway_proxy);
    let body = serde_json::json!({"path": path});
    let response = client
        .post(&url)
        .json(&body)
        .send()
        .await
        .context("Failed to send get catalog request")?;

    if response.status().is_success() {
        let result: serde_json::Value = response.json().await?;
        println!("{}", serde_json::to_string_pretty(&result)?);
    } else {
        let status = response.status();
        let text = response.text().await?;
        eprintln!("Failed to get catalog resource: {} - {}", status, text);
    }
    Ok(())
}

async fn get_credential(
    client: &Client,
    base_url: &str,
    use_gateway_proxy: bool,
    name: &str,
    include_data: bool,
) -> Result<()> {
    let url = api_url(
        base_url,
        &format!("credentials/{}?include_data={}", name, include_data),
        use_gateway_proxy,
    );
    let response = client
        .get(&url)
        .send()
        .await
        .context("Failed to send get credential request")?;

    if response.status().is_success() {
        let result: serde_json::Value = response.json().await?;
        println!("{}", serde_json::to_string_pretty(&result)?);
    } else {
        let status = response.status();
        let text = response.text().await?;
        eprintln!("Failed to get credential: {} - {}", status, text);
    }
    Ok(())
}

async fn execute_query(
    client: &Client,
    base_url: &str,
    use_gateway_proxy: bool,
    query: &str,
    schema: &str,
    format: &str,
) -> Result<()> {
    let url = api_url(base_url, "postgres/execute", use_gateway_proxy);

    let payload = serde_json::json!({
        "query": query,
        "schema": schema
    });

    let response = client
        .post(&url)
        .header("Content-Type", "application/json")
        .json(&payload)
        .send()
        .await
        .context("Failed to send query request")?;

    if response.status().is_success() {
        let result: serde_json::Value = response.json().await?;

        match format {
            "json" => {
                // Pretty print JSON
                println!("{}", serde_json::to_string_pretty(&result)?);
            }
            "table" => {
                // Extract column names from query
                let column_names = extract_column_names(query);
                // Format as table
                format_as_table(&result, &column_names)?;
            }
            _ => {
                eprintln!("Unknown format: {}. Use 'table' or 'json'", format);
                std::process::exit(1);
            }
        }
    } else {
        let status = response.status();
        let text = response.text().await?;
        eprintln!("Failed to execute query: {} - {}", status, text);
        std::process::exit(1);
    }

    Ok(())
}

fn extract_column_names(query: &str) -> Vec<String> {
    // Simple column name extraction from SELECT query
    let query_upper = query.to_uppercase();

    // Find SELECT and FROM positions
    if let Some(select_pos) = query_upper.find("SELECT") {
        let after_select = &query[select_pos + 6..].trim_start();

        // Find FROM keyword
        let from_pos = after_select.to_uppercase().find(" FROM");
        let columns_str = if let Some(pos) = from_pos {
            &after_select[..pos]
        } else {
            after_select
        };

        // Split by comma and clean up
        let columns: Vec<String> = columns_str
            .split(',')
            .map(|s| {
                let s = s.trim();
                // Handle aliases (AS keyword)
                if let Some(as_pos) = s.to_uppercase().rfind(" AS ") {
                    s[as_pos + 4..].trim().to_string()
                } else {
                    // Get the last part after dot (for qualified names like table.column)
                    s.split('.').last().unwrap_or(s).trim().to_string()
                }
            })
            .collect();

        columns
    } else {
        Vec::new()
    }
}

fn format_as_table(result: &serde_json::Value, column_names: &[String]) -> Result<()> {
    // Check if result has the expected structure
    if let Some(result_array) = result.get("result").and_then(|r| r.as_array()) {
        if result_array.is_empty() {
            println!("(0 rows)");
            return Ok(());
        }

        // The API returns result as array of arrays: [[val1, val2], [val3, val4]]
        // We need to determine column count from first row
        let first_row = &result_array[0];
        if let Some(row_array) = first_row.as_array() {
            let col_count = row_array.len();

            // Use extracted column names if available and count matches, otherwise fall back to generic
            let columns: Vec<String> = if column_names.len() == col_count {
                column_names.to_vec()
            } else {
                (1..=col_count).map(|i| format!("column_{}", i)).collect()
            };

            // Calculate column widths
            let mut col_widths: Vec<usize> = columns.iter().map(|c| c.len()).collect();

            for row in result_array {
                if let Some(row_array) = row.as_array() {
                    for (i, val) in row_array.iter().enumerate() {
                        if i < col_widths.len() {
                            let val_str = format_value(val);
                            col_widths[i] = col_widths[i].max(val_str.len());
                        }
                    }
                }
            }

            // Print header
            print!("┌");
            for (i, width) in col_widths.iter().enumerate() {
                print!("{}", "─".repeat(width + 2));
                if i < col_widths.len() - 1 {
                    print!("┬");
                }
            }
            println!("┐");

            print!("│");
            for (i, col) in columns.iter().enumerate() {
                let width = col_widths.get(i).copied().unwrap_or(col.len());
                print!(" {:<width$} │", col, width = width);
            }
            println!();

            print!("├");
            for (i, width) in col_widths.iter().enumerate() {
                print!("{}", "─".repeat(width + 2));
                if i < col_widths.len() - 1 {
                    print!("┼");
                }
            }
            println!("┤");

            // Print rows
            for row in result_array {
                if let Some(row_array) = row.as_array() {
                    print!("│");
                    for (i, val) in row_array.iter().enumerate() {
                        if i < col_widths.len() {
                            let val_str = format_value(val);
                            let width = col_widths[i];
                            print!(" {:<width$} │", val_str, width = width);
                        }
                    }
                    println!();
                }
            }

            print!("└");
            for (i, width) in col_widths.iter().enumerate() {
                print!("{}", "─".repeat(width + 2));
                if i < col_widths.len() - 1 {
                    print!("┴");
                }
            }
            println!("┘");

            println!("({} rows)", result_array.len());
        }
    } else {
        // Fallback to pretty JSON if structure doesn't match
        println!("{}", serde_json::to_string_pretty(result)?);
    }

    Ok(())
}

fn format_value(val: &serde_json::Value) -> String {
    match val {
        serde_json::Value::Null => "NULL".to_string(),
        serde_json::Value::Bool(b) => b.to_string(),
        serde_json::Value::Number(n) => n.to_string(),
        serde_json::Value::String(s) => s.clone(),
        serde_json::Value::Array(_) | serde_json::Value::Object(_) => {
            serde_json::to_string(val).unwrap_or_else(|_| "{}".to_string())
        }
    }
}

async fn run_tui(base_url: &str, client: Client, use_gateway_proxy: bool) -> Result<()> {
    enable_raw_mode()?;
    let mut stdout = io::stdout();
    execute!(stdout, EnterAlternateScreen, EnableMouseCapture)?;
    let backend = CrosstermBackend::new(stdout);
    let mut terminal = Terminal::new(backend)?;

    let app = App::new(base_url, client, use_gateway_proxy);
    let res = run_app(&mut terminal, app).await;

    disable_raw_mode()?;
    execute!(terminal.backend_mut(), LeaveAlternateScreen, DisableMouseCapture)?;
    terminal.show_cursor()?;

    if let Err(err) = res {
        println!("{:?}", err)
    }

    Ok(())
}

struct App {
    base_url: String,
    use_gateway_proxy: bool,
    playbooks: Vec<String>,
    state: ListState,
    client: Client,
}

impl App {
    fn new(base_url: &str, client: Client, use_gateway_proxy: bool) -> App {
        App {
            base_url: base_url.to_string(),
            use_gateway_proxy,
            playbooks: Vec::new(),
            state: ListState::default(),
            client,
        }
    }

    async fn update_playbooks(&mut self) -> Result<()> {
        let url = api_url(&self.base_url, "catalog/list/Playbook", self.use_gateway_proxy);
        let response = self.client.get(&url).send().await?;
        if response.status().is_success() {
            let json: serde_json::Value = response.json().await?;
            if let Some(list) = json.as_array() {
                self.playbooks = list.iter().filter_map(|v| v.as_str().map(|s| s.to_string())).collect();
            } else if let Some(obj) = json.as_object() {
                // Sometimes it might be an object where keys are paths
                self.playbooks = obj.keys().cloned().collect();
            }
        }
        Ok(())
    }

    fn next(&mut self) {
        let i = match self.state.selected() {
            Some(i) => {
                if i >= self.playbooks.len() - 1 {
                    0
                } else {
                    i + 1
                }
            }
            None => 0,
        };
        self.state.select(Some(i));
    }

    fn previous(&mut self) {
        let i = match self.state.selected() {
            Some(i) => {
                if i == 0 {
                    self.playbooks.len() - 1
                } else {
                    i - 1
                }
            }
            None => 0,
        };
        self.state.select(Some(i));
    }
}

async fn run_app<B: Backend>(terminal: &mut Terminal<B>, mut app: App) -> Result<()> {
    app.update_playbooks().await.ok();
    let tick_rate = Duration::from_millis(250);
    let mut last_tick = Instant::now();

    loop {
        terminal.draw(|f| ui(f, &mut app))?;

        let timeout = tick_rate
            .checked_sub(last_tick.elapsed())
            .unwrap_or_else(|| Duration::from_secs(0));

        if crossterm::event::poll(timeout)? {
            if let Event::Key(key) = event::read()? {
                match key.code {
                    KeyCode::Char('q') => return Ok(()),
                    KeyCode::Down | KeyCode::Char('j') => app.next(),
                    KeyCode::Up | KeyCode::Char('k') => app.previous(),
                    KeyCode::Char('r') => {
                        app.update_playbooks().await.ok();
                    }
                    _ => {}
                }
            }
        }
        if last_tick.elapsed() >= tick_rate {
            last_tick = Instant::now();
        }
    }
}

fn ui(f: &mut Frame, app: &mut App) {
    let chunks = Layout::default()
        .direction(Direction::Vertical)
        .constraints([Constraint::Length(3), Constraint::Min(0), Constraint::Length(3)].as_ref())
        .split(f.size());

    let header =
        Paragraph::new("NoETL Control (noetl) - Playbooks").block(Block::default().borders(Borders::ALL).title("Info"));
    f.render_widget(header, chunks[0]);

    let items: Vec<ListItem> = app.playbooks.iter().map(|i| ListItem::new(i.as_str())).collect();

    let list = List::new(items)
        .block(Block::default().borders(Borders::ALL).title("Playbooks"))
        .highlight_symbol(">> ");

    f.render_stateful_widget(list, chunks[1], &mut app.state);

    let footer = Paragraph::new("q: Quit | r: Refresh | ↑/↓: Navigate")
        .block(Block::default().borders(Borders::ALL).title("Help"));
    f.render_widget(footer, chunks[2]);
}

// ============================================================================
// Server Management
// ============================================================================

async fn start_server(init_db: bool) -> Result<()> {
    use std::net::TcpStream;
    use std::process::{Command, Stdio};

    let pid_dir = dirs::home_dir()
        .context("Could not determine home directory")?
        .join(".noetl");
    std::fs::create_dir_all(&pid_dir)?;

    let pid_file = pid_dir.join("noetl_server.pid");

    // Check if server is already running
    if pid_file.exists() {
        let pid_str = std::fs::read_to_string(&pid_file)?;
        if let Ok(pid) = pid_str.trim().parse::<i32>() {
            if process_exists(pid) {
                println!(
                    "Server already running with PID {} (PID file: {})",
                    pid,
                    pid_file.display()
                );
                println!("Use 'noetl server stop' to stop it first.");
                return Ok(());
            } else {
                println!("Found stale PID file. Removing it.");
                std::fs::remove_file(&pid_file)?;
            }
        }
    }

    // Get server configuration from environment
    let host = std::env::var("NOETL_HOST").unwrap_or_else(|_| "0.0.0.0".to_string());
    let port = std::env::var("NOETL_PORT").unwrap_or_else(|_| "8082".to_string());

    // Check port availability
    if let Ok(_stream) = TcpStream::connect(format!(
        "{}:{}",
        if host == "0.0.0.0" { "127.0.0.1" } else { &host },
        port
    )) {
        eprintln!("Error: Port {}:{} is already in use", host, port);
        return Err(anyhow::anyhow!("Port already in use"));
    }

    println!("Starting NoETL server at http://{}:{}...", host, port);

    // Spawn Python server subprocess using new entry point
    let mut cmd = Command::new("python");
    cmd.args(&["-m", "noetl.server"])
        .arg("--host")
        .arg(&host)
        .arg("--port")
        .arg(&port);

    if init_db {
        cmd.arg("--init-db");
    }

    cmd.stdout(Stdio::null()).stderr(Stdio::null());

    // Set environment variables
    if let Ok(val) = std::env::var("NOETL_ENABLE_UI") {
        cmd.env("NOETL_ENABLE_UI", val);
    }
    if let Ok(val) = std::env::var("NOETL_DEBUG") {
        cmd.env("NOETL_DEBUG", val);
    }

    let child = cmd
        .spawn()
        .context("Failed to spawn server process. Is Python and noetl package installed?")?;

    let pid = child.id();

    // Write PID file
    std::fs::write(&pid_file, pid.to_string())?;
    println!("Server started with PID {}", pid);
    println!("PID file: {}", pid_file.display());

    // Optional: Initialize database
    if init_db {
        println!("Waiting for server to be ready...");
        tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;

        let client = Client::new();
        let base_url = format!("http://localhost:{}", port);

        println!("Initializing database schema...");
        let response = client.post(&format!("{}/api/db/init", base_url)).send().await;

        match response {
            Ok(resp) if resp.status().is_success() => {
                println!("Database initialized successfully.");
            }
            Ok(resp) => {
                eprintln!("Warning: Database initialization returned status {}", resp.status());
            }
            Err(e) => {
                eprintln!("Warning: Could not initialize database: {}", e);
            }
        }
    }

    Ok(())
}

async fn stop_server(force: bool) -> Result<()> {
    let pid_dir = dirs::home_dir()
        .context("Could not determine home directory")?
        .join(".noetl");
    let pid_file = pid_dir.join("noetl_server.pid");

    if !pid_file.exists() {
        println!("No running NoETL server found (no PID file at {}).", pid_file.display());
        return Ok(());
    }

    let pid_str = std::fs::read_to_string(&pid_file)?;
    let pid: i32 = pid_str.trim().parse().context("Invalid PID in file")?;

    if !process_exists(pid) {
        println!("Process {} not found. The server may have been stopped already.", pid);
        std::fs::remove_file(&pid_file)?;
        return Ok(());
    }

    if !force {
        print!("Stop NoETL server with PID {}? [y/N]: ", pid);
        std::io::Write::flush(&mut std::io::stdout())?;

        let mut input = String::new();
        std::io::stdin().read_line(&mut input)?;

        if !input.trim().eq_ignore_ascii_case("y") {
            println!("Operation cancelled.");
            return Ok(());
        }
    }

    println!("Stopping NoETL server with PID {}...", pid);

    // Send SIGTERM
    send_signal(pid, nix::sys::signal::Signal::SIGTERM)?;

    // Wait for graceful shutdown (10 seconds)
    for _ in 0..20 {
        if !process_exists(pid) {
            std::fs::remove_file(&pid_file)?;
            println!("NoETL server stopped successfully.");
            return Ok(());
        }
        tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
    }

    // Force kill if still running
    if force {
        println!("Server didn't stop gracefully. Force killing...");
        send_signal(pid, nix::sys::signal::Signal::SIGKILL)?;
    } else {
        print!("Server didn't stop gracefully. Force kill? [y/N]: ");
        std::io::Write::flush(&mut std::io::stdout())?;

        let mut input = String::new();
        std::io::stdin().read_line(&mut input)?;

        if input.trim().eq_ignore_ascii_case("y") {
            println!("Force killing NoETL server with PID {}...", pid);
            send_signal(pid, nix::sys::signal::Signal::SIGKILL)?;
        }
    }

    std::fs::remove_file(&pid_file)?;
    println!("NoETL server stopped.");

    Ok(())
}

// ============================================================================
// Worker Management
// ============================================================================

async fn start_worker(_max_workers: Option<usize>) -> Result<()> {
    use std::process::{Command, Stdio};

    let pid_dir = dirs::home_dir()
        .context("Could not determine home directory")?
        .join(".noetl");
    std::fs::create_dir_all(&pid_dir)?;

    // Determine worker name based on pool config
    let worker_name = std::env::var("NOETL_WORKER_POOL_NAME")
        .unwrap_or_else(|_| "default".to_string())
        .replace("-", "_");

    let pid_file = pid_dir.join(format!("noetl_worker_{}.pid", worker_name));

    // Check if worker is already running
    if pid_file.exists() {
        let pid_str = std::fs::read_to_string(&pid_file)?;
        if let Ok(pid) = pid_str.trim().parse::<i32>() {
            if process_exists(pid) {
                println!("Worker '{}' already running with PID {}", worker_name, pid);
                println!("Use 'noetl worker stop --name {}' to stop it first.", worker_name);
                return Ok(());
            } else {
                println!("Found stale PID file. Removing it.");
                std::fs::remove_file(&pid_file)?;
            }
        }
    }

    println!("Starting NoETL worker '{}' (v2 architecture)...", worker_name);

    // Build Python worker command - execute worker module directly
    // python -m noetl.worker starts V2 worker via __main__.py
    let mut cmd = Command::new("python");
    cmd.args(&["-m", "noetl.worker"]);

    cmd.stdout(Stdio::null()).stderr(Stdio::null());

    let child = cmd
        .spawn()
        .context("Failed to spawn worker process. Is Python and noetl package installed?")?;

    let pid = child.id();

    // Write PID file
    std::fs::write(&pid_file, pid.to_string())?;
    println!("Worker '{}' started with PID {}", worker_name, pid);
    println!("PID file: {}", pid_file.display());

    Ok(())
}

async fn stop_worker(name: Option<String>, force: bool) -> Result<()> {
    use std::io::Write;

    let pid_dir = dirs::home_dir()
        .context("Could not determine home directory")?
        .join(".noetl");

    // If no name provided, list workers and prompt
    let pid_file = if let Some(worker_name) = name {
        let normalized_name = worker_name.replace("-", "_");
        pid_dir.join(format!("noetl_worker_{}.pid", normalized_name))
    } else {
        // List all worker PID files
        let entries = std::fs::read_dir(&pid_dir)?
            .filter_map(|e| e.ok())
            .filter(|e| {
                e.file_name().to_string_lossy().starts_with("noetl_worker_")
                    && e.file_name().to_string_lossy().ends_with(".pid")
            })
            .collect::<Vec<_>>();

        if entries.is_empty() {
            println!("No running NoETL worker services found.");
            return Ok(());
        }

        println!("Running workers:");
        for (i, entry) in entries.iter().enumerate() {
            let worker_name = entry
                .file_name()
                .to_string_lossy()
                .strip_prefix("noetl_worker_")
                .and_then(|s| s.strip_suffix(".pid"))
                .unwrap_or("unknown")
                .to_string();

            if let Ok(pid_str) = std::fs::read_to_string(entry.path()) {
                println!("  {}. {} (PID: {})", i + 1, worker_name, pid_str.trim());
            } else {
                println!("  {}. {} (PID file corrupted)", i + 1, worker_name);
            }
        }

        if entries.len() == 1 {
            entries[0].path()
        } else {
            print!("Enter the number of the worker to stop: ");
            std::io::stdout().flush()?;

            let mut input = String::new();
            std::io::stdin().read_line(&mut input)?;

            let choice: usize = input.trim().parse().context("Invalid number")?;

            if choice < 1 || choice > entries.len() {
                return Err(anyhow::anyhow!("Invalid choice"));
            }

            entries[choice - 1].path()
        }
    };

    if !pid_file.exists() {
        println!("Worker PID file not found: {}", pid_file.display());
        return Ok(());
    }

    let pid_str = std::fs::read_to_string(&pid_file)?;
    let pid: i32 = pid_str.trim().parse().context("Invalid PID in file")?;

    if !process_exists(pid) {
        println!("Process {} not found. The worker may have been stopped already.", pid);
        std::fs::remove_file(&pid_file)?;
        return Ok(());
    }

    if !force {
        print!("Stop NoETL worker with PID {}? [y/N]: ", pid);
        std::io::Write::flush(&mut std::io::stdout())?;

        let mut input = String::new();
        std::io::stdin().read_line(&mut input)?;

        if !input.trim().eq_ignore_ascii_case("y") {
            println!("Operation cancelled.");
            return Ok(());
        }
    }

    println!("Stopping NoETL worker with PID {}...", pid);

    // Send SIGTERM
    send_signal(pid, nix::sys::signal::Signal::SIGTERM)?;

    // Wait for graceful shutdown (10 seconds)
    for _ in 0..20 {
        if !process_exists(pid) {
            std::fs::remove_file(&pid_file)?;
            println!("NoETL worker stopped successfully.");
            return Ok(());
        }
        tokio::time::sleep(tokio::time::Duration::from_millis(500)).await;
    }

    // Force kill if still running
    if force {
        println!("Worker didn't stop gracefully. Force killing...");
        send_signal(pid, nix::sys::signal::Signal::SIGKILL)?;
    } else {
        print!("Worker didn't stop gracefully. Force kill? [y/N]: ");
        std::io::Write::flush(&mut std::io::stdout())?;

        let mut input = String::new();
        std::io::stdin().read_line(&mut input)?;

        if input.trim().eq_ignore_ascii_case("y") {
            println!("Force killing NoETL worker with PID {}...", pid);
            send_signal(pid, nix::sys::signal::Signal::SIGKILL)?;
        }
    }

    std::fs::remove_file(&pid_file)?;
    println!("NoETL worker stopped.");

    Ok(())
}

// ============================================================================
// Database Management
// ============================================================================

async fn db_init(client: &Client, base_url: &str, use_gateway_proxy: bool) -> Result<()> {
    println!("Initializing NoETL database schema...");

    let url = api_url(base_url, "db/init", use_gateway_proxy);
    let response = client
        .post(&url)
        .send()
        .await
        .context("Failed to send database init request")?;

    if response.status().is_success() {
        let result: serde_json::Value = response.json().await?;
        println!("Database initialized successfully:");
        println!("{}", serde_json::to_string_pretty(&result)?);
    } else {
        let status = response.status();
        let text = response.text().await?;
        eprintln!("Failed to initialize database: {} - {}", status, text);
        return Err(anyhow::anyhow!("Database initialization failed"));
    }

    Ok(())
}

async fn db_validate(client: &Client, base_url: &str, use_gateway_proxy: bool) -> Result<()> {
    println!("Validating NoETL database schema...");

    let url = api_url(base_url, "db/validate", use_gateway_proxy);
    let response = client
        .get(&url)
        .send()
        .await
        .context("Failed to send database validate request")?;

    if response.status().is_success() {
        let result: serde_json::Value = response.json().await?;
        println!("Database validation result:");
        println!("{}", serde_json::to_string_pretty(&result)?);
    } else {
        let status = response.status();
        let text = response.text().await?;
        eprintln!("Failed to validate database: {} - {}", status, text);
        return Err(anyhow::anyhow!("Database validation failed"));
    }

    Ok(())
}

// ============================================================================
// Helper Functions for Process Management
// ============================================================================

fn process_exists(pid: i32) -> bool {
    use sysinfo::{ProcessesToUpdate, System};

    let mut system = System::new_all();
    system.refresh_processes(ProcessesToUpdate::All);

    system.process(sysinfo::Pid::from(pid as usize)).is_some()
}

fn send_signal(pid: i32, signal: nix::sys::signal::Signal) -> Result<()> {
    use nix::sys::signal::kill;
    use nix::unistd::Pid;

    kill(Pid::from_raw(pid), signal).context("Failed to send signal to process")?;

    Ok(())
}

// ============================================================================
// Build Commands
// ============================================================================

async fn build_docker_image(no_cache: bool, platform: Option<String>) -> Result<()> {
    use chrono::Local;
    use std::io::{BufRead, BufReader};
    use std::process::Command;

    let registry = "local";
    let image_name = "noetl";
    let image_tag = Local::now().format("%Y-%m-%d-%H-%M").to_string();

    // Default to host platform if not specified (prevents slow QEMU emulation on ARM)
    let target_platform = platform;

    println!("Building Docker image: {}/{}:{}", registry, image_name, image_tag);
    if let Some(ref p) = target_platform {
        println!("Target platform: {}", p);
    } else {
        println!("Target platform: native");
    }

    let mut cmd = Command::new("docker");
    cmd.arg("buildx");
    cmd.arg("build");

    if no_cache {
        cmd.arg("--no-cache");
    }
    cmd.arg("--progress=plain");

    // Build for specified platform if provided
    if let Some(ref p) = target_platform {
        cmd.arg("--platform").arg(p);
    }

    // Load image to local Docker daemon (required for kind load)
    cmd.arg("--load");

    cmd.arg("-t")
        .arg(format!("{}/{}:{}", registry, image_name, image_tag))
        .arg("-f")
        .arg("docker/noetl/dev/Dockerfile")
        .arg(".")
        .stdout(std::process::Stdio::piped())
        .stderr(std::process::Stdio::piped());

    let cr = std::env::current_dir()?;

    cmd.current_dir(cr);

    println!("cwd = {:?}", std::env::current_dir()?);

    // cmd.env("DOCKER_BUILDKIT", "0");

    println!(
        "Running: docker buildx build{} --progress=plain{} -t {}/{}:{} -f docker/noetl/dev/Dockerfile .",
        if no_cache { " --no-cache" } else { "" },
        if let Some(ref p) = target_platform {
            format!(" --platform {}", p)
        } else {
            "".to_string()
        },
        registry,
        image_name,
        image_tag
    );

    let mut child = cmd.spawn().context("Failed to spawn docker build command")?;

    // Clone the stdout and stderr to read in separate threads
    let stdout = child.stdout.take().unwrap();
    let stderr = child.stderr.take().unwrap();

    let stdout_thread = std::thread::spawn(move || {
        let reader = BufReader::new(stdout);
        for line in reader.lines() {
            if let Ok(line) = line {
                println!("{}", line);
            }
        }
    });

    let stderr_thread = std::thread::spawn(move || {
        let reader = BufReader::new(stderr);
        for line in reader.lines() {
            if let Ok(line) = line {
                println!("{}", line);
            }
        }
    });

    // Wait for both threads to finish
    stdout_thread.join().unwrap();
    stderr_thread.join().unwrap();

    let status = child.wait()?;

    if !status.success() {
        return Err(anyhow::anyhow!("Docker build failed with status: {}", status));
    }

    // Save image tag to file
    std::fs::write(".noetl_last_build_tag.txt", &image_tag)?;
    println!("✓ Image built successfully: {}/{}:{}", registry, image_name, image_tag);
    println!("✓ Tag saved to .noetl_last_build_tag.txt");

    Ok(())
}

// ============================================================================
// Kubernetes Commands
// ============================================================================

async fn k8s_deploy() -> Result<()> {
    println!("Deploying NoETL to Kubernetes...");

    // Read image tag
    let image_tag = std::fs::read_to_string(".noetl_last_build_tag.txt")
        .context("Failed to read .noetl_last_build_tag.txt - have you built the image?")?;
    let image_tag = image_tag.trim();

    let registry = "local";
    let image_name = "noetl";
    let full_image = format!("{}/{}:{}", registry, image_name, image_tag);

    println!("Using image: {}", full_image);

    // Load image to kind cluster
    println!("Loading image to kind cluster...");
    run_command(&["kind", "load", "docker-image", &full_image, "--name", "noetl"])?;

    // Set kubectl context
    println!("Setting kubectl context to kind-noetl...");
    run_command(&["kubectl", "config", "use-context", "kind-noetl"])?;

    // Apply namespace
    println!("Creating namespace...");
    run_command(&["kubectl", "apply", "-f", "ci/manifests/noetl/namespace/namespace.yaml"])?;
    tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;

    // Update image in deployment files using yq
    println!("Updating deployment image references...");
    update_deployment_image("ci/manifests/noetl/server-deployment.yaml", &full_image)?;
    update_deployment_image("ci/manifests/noetl/worker-deployment.yaml", &full_image)?;

    // Apply manifests
    println!("Applying Kubernetes manifests...");
    run_command(&["kubectl", "apply", "-f", "ci/manifests/noetl/"])?;

    // Restore original image placeholders
    println!("Restoring deployment templates...");
    update_deployment_image("ci/manifests/noetl/server-deployment.yaml", "image_name:image_tag")?;
    update_deployment_image("ci/manifests/noetl/worker-deployment.yaml", "image_name:image_tag")?;

    println!("✓ NoETL deployed successfully");
    println!("  UI:  http://localhost:8082");
    println!("  API: http://localhost:8082/docs");

    Ok(())
}

async fn k8s_remove() -> Result<()> {
    println!("Removing NoETL from Kubernetes...");

    // Set kubectl context
    run_command(&["kubectl", "config", "use-context", "kind-noetl"])?;

    // Delete manifests
    run_command(&["kubectl", "delete", "-f", "ci/manifests/noetl/"])?;

    println!("✓ NoETL removed successfully");

    Ok(())
}

async fn k8s_redeploy(no_cache: bool, platform: Option<String>) -> Result<()> {
    println!("Rebuilding and redeploying NoETL...");

    // Build image
    build_docker_image(no_cache, platform.clone()).await?;

    // Remove existing deployment
    k8s_remove().await.ok(); // Ignore errors if not deployed

    // Load image to kind
    let image_tag = std::fs::read_to_string(".noetl_last_build_tag.txt")?.trim().to_string();
    println!("Loading image to kind cluster...");
    run_command(&[
        "kind",
        "load",
        "docker-image",
        &format!("local/noetl:{}", image_tag),
        "--name",
        "noetl",
    ])?;

    // Deploy
    k8s_deploy().await?;

    println!("✓ NoETL redeployed successfully");

    Ok(())
}

async fn k8s_reset(no_cache: bool, platform: Option<String>) -> Result<()> {
    println!("Resetting NoETL (full rebuild + schema reset + test setup)...");

    // Reset postgres schema
    println!("Resetting Postgres schema...");
    run_command(&["task", "postgres:k8s:schema-reset"])?;

    // Redeploy
    k8s_redeploy(no_cache, platform).await?;

    // Install noetl CLI with dev extras
    println!("Installing NoETL CLI with dev dependencies...");
    run_command(&["uv", "pip", "install", "-e", ".[dev]"])?;

    // Wait for deployment to be ready
    println!("Waiting for deployment to be ready...");
    tokio::time::sleep(tokio::time::Duration::from_secs(30)).await;

    // Setup test environment
    println!("Setting up test environment...");
    run_command(&["task", "test:k8s:create-tables"])?;
    run_command(&["task", "test:k8s:register-credentials"])?;

    println!("✓ NoETL reset complete");
    println!("  UI:  http://localhost:8082");
    println!("  API: http://localhost:8082/docs");

    Ok(())
}

// Helper functions for k8s commands
fn run_command(args: &[&str]) -> Result<()> {
    use std::process::Command;

    let output = Command::new(args[0])
        .args(&args[1..])
        .output()
        .context(format!("Failed to execute: {}", args.join(" ")))?;

    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        return Err(anyhow::anyhow!("Command failed: {}\n{}", args.join(" "), stderr));
    }

    print!("{}", String::from_utf8_lossy(&output.stdout));

    Ok(())
}

fn update_deployment_image(file_path: &str, image: &str) -> Result<()> {
    use std::process::Command;

    // Check which yq version is installed
    let yq_version = Command::new("yq").arg("--version").output();

    let is_mikefarah = if let Ok(output) = yq_version {
        String::from_utf8_lossy(&output.stdout).contains("mikefarah")
            || String::from_utf8_lossy(&output.stderr).contains("mikefarah")
    } else {
        false
    };

    if is_mikefarah {
        // mikefarah/yq (v4+)
        Command::new("yq")
            .arg("-i")
            .arg(format!(".spec.template.spec.containers[0].image = \"{}\"", image))
            .arg(file_path)
            .output()
            .context("Failed to update deployment with yq")?;
    } else {
        // kislyuk/yq (python version)
        let temp_file = format!("{}.tmp", file_path);
        Command::new("yq")
            .arg("-y")
            .arg(format!(".spec.template.spec.containers[0].image = \"{}\"", image))
            .arg(file_path)
            .stdout(std::fs::File::create(&temp_file)?)
            .output()
            .context("Failed to update deployment with yq")?;

        std::fs::rename(&temp_file, file_path)?;
    }

    Ok(())
}

// =============================================================================
// Infrastructure as Playbook (IaP) Command Handlers
// =============================================================================

async fn handle_iap_command(command: IapCommand) -> Result<()> {
    match command {
        IapCommand::Init {
            project,
            bucket,
            region,
            state_db,
            workspace,
            state_path,
        } => iap_init(&project, bucket.as_deref(), &region, &state_db, &workspace, &state_path).await,
        IapCommand::Plan {
            playbook,
            variables,
            verbose,
        } => iap_plan(&playbook, &variables, verbose).await,
        IapCommand::Apply {
            playbook,
            variables,
            auto_approve,
            verbose,
        } => iap_apply(&playbook, &variables, auto_approve, verbose).await,
        IapCommand::State { command } => handle_iap_state_command(command).await,
        IapCommand::Sync { command } => handle_iap_sync_command(command).await,
        IapCommand::Drift { command } => handle_iap_drift_command(command).await,
        IapCommand::Workspace { command } => handle_iap_workspace_command(command).await,
    }
}

async fn handle_iap_state_command(command: IapStateCommand) -> Result<()> {
    match command {
        IapStateCommand::List { resource_type, format } => iap_state_list(resource_type.as_deref(), &format).await,
        IapStateCommand::Show { resource } => iap_state_show(&resource).await,
        IapStateCommand::Rm { resource, force } => iap_state_rm(&resource, force).await,
        IapStateCommand::Query { sql } => iap_state_query(&sql).await,
    }
}

async fn handle_iap_sync_command(command: IapSyncCommand) -> Result<()> {
    match command {
        IapSyncCommand::Push { force } => iap_sync_push(force).await,
        IapSyncCommand::Pull { force } => iap_sync_pull(force).await,
        IapSyncCommand::Status => iap_sync_status().await,
    }
}

async fn handle_iap_drift_command(command: IapDriftCommand) -> Result<()> {
    match command {
        IapDriftCommand::Detect {
            resource_type,
            resource,
        } => iap_drift_detect(resource_type.as_deref(), resource.as_deref()).await,
        IapDriftCommand::Report { format } => iap_drift_report(&format).await,
    }
}

async fn handle_iap_workspace_command(command: IapWorkspaceCommand) -> Result<()> {
    match command {
        IapWorkspaceCommand::List { remote } => iap_workspace_list(remote).await,
        IapWorkspaceCommand::Switch { name, pull } => iap_workspace_switch(&name, pull).await,
        IapWorkspaceCommand::Current => iap_workspace_current().await,
        IapWorkspaceCommand::Create { name, from, switch } => {
            iap_workspace_create(&name, from.as_deref(), switch).await
        }
        IapWorkspaceCommand::Delete { name, remote, force } => iap_workspace_delete(&name, remote, force).await,
    }
}

/// Initialize IaP state database and configuration
async fn iap_init(
    project: &str,
    bucket: Option<&str>,
    region: &str,
    state_db: &str,
    workspace: &str,
    state_path_template: &str,
) -> Result<()> {
    use duckdb::{params, Connection};

    println!("Initializing IaP for project: {}", project);
    println!("  Region: {}", region);
    println!("  Workspace: {}", workspace);
    println!("  State DB: {}", state_db);

    // Create state directory
    let state_path = PathBuf::from(state_db);
    if let Some(parent) = state_path.parent() {
        fs::create_dir_all(parent)?;
    }

    // Open/create DuckDB database
    let conn = Connection::open(&state_path).context("Failed to create state database")?;

    // Create schema tables
    conn.execute_batch(
        r#"
        -- IaP Configuration
        CREATE TABLE IF NOT EXISTS iap_config (
            key VARCHAR PRIMARY KEY,
            value VARCHAR NOT NULL,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );

        -- Resources table - tracks all managed infrastructure
        CREATE TABLE IF NOT EXISTS resources (
            resource_id VARCHAR PRIMARY KEY,
            resource_type VARCHAR NOT NULL,
            resource_name VARCHAR NOT NULL,
            provider VARCHAR NOT NULL DEFAULT 'gcp',
            project VARCHAR,
            region VARCHAR,
            zone VARCHAR,
            status VARCHAR NOT NULL DEFAULT 'pending',
            config JSON,
            state JSON,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            last_sync_at TIMESTAMP,
            drift_detected BOOLEAN DEFAULT FALSE
        );

        -- Execution history
        CREATE TABLE IF NOT EXISTS executions (
            execution_id INTEGER PRIMARY KEY,
            playbook_path VARCHAR NOT NULL,
            action VARCHAR NOT NULL,
            status VARCHAR NOT NULL,
            started_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            completed_at TIMESTAMP,
            error_message VARCHAR,
            changes JSON
        );

        -- Create sequence for execution_id
        CREATE SEQUENCE IF NOT EXISTS exec_seq START 1;

        -- Dependencies between resources
        CREATE TABLE IF NOT EXISTS dependencies (
            source_id VARCHAR NOT NULL,
            target_id VARCHAR NOT NULL,
            dependency_type VARCHAR DEFAULT 'requires',
            PRIMARY KEY (source_id, target_id)
        );

        -- Drift detection results
        CREATE TABLE IF NOT EXISTS drift_log (
            id INTEGER PRIMARY KEY,
            resource_id VARCHAR NOT NULL,
            detected_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            drift_type VARCHAR NOT NULL,
            expected_value JSON,
            actual_value JSON,
            resolved_at TIMESTAMP,
            resolved_by VARCHAR
        );
        "#,
    )
    .context("Failed to create schema tables")?;

    // Store configuration
    conn.execute(
        "INSERT OR REPLACE INTO iap_config (key, value) VALUES ('project', ?)",
        params![project],
    )?;
    conn.execute(
        "INSERT OR REPLACE INTO iap_config (key, value) VALUES ('region', ?)",
        params![region],
    )?;
    conn.execute(
        "INSERT OR REPLACE INTO iap_config (key, value) VALUES ('workspace', ?)",
        params![workspace],
    )?;
    conn.execute(
        "INSERT OR REPLACE INTO iap_config (key, value) VALUES ('state_path_template', ?)",
        params![state_path_template],
    )?;

    if let Some(b) = bucket {
        conn.execute(
            "INSERT OR REPLACE INTO iap_config (key, value) VALUES ('bucket', ?)",
            params![b],
        )?;
        println!("  GCS Bucket: {}", b);

        // Show the actual remote path that will be used
        let remote_path = state_path_template.replace("{workspace}", workspace);
        println!("  Remote Path: gs://{}/{}", b, remote_path);
    }

    // Register workspace in registry for switching
    let mut registry = load_workspace_registry()?;
    let remote_path = bucket.map(|b| {
        let path = state_path_template.replace("{workspace}", workspace);
        format!("gs://{}/{}", b, path)
    });
    let entry = WorkspaceEntry {
        name: workspace.to_string(),
        project: project.to_string(),
        region: region.to_string(),
        bucket: bucket.map(|b| b.to_string()),
        state_path_template: state_path_template.to_string(),
        remote_path,
        created_at: chrono::Utc::now().to_rfc3339(),
        last_used: Some(chrono::Utc::now().to_rfc3339()),
    };
    registry.insert(workspace.to_string(), entry);
    save_workspace_registry(&registry)?;

    println!("\nIaP initialized successfully.");
    println!("State database created at: {}", state_db);

    if bucket.is_some() {
        println!("\nTo sync state with GCS, run:");
        println!("  noetl iap sync push");
    }

    Ok(())
}

/// Plan infrastructure changes (dry-run)
async fn iap_plan(playbook: &PathBuf, variables: &[String], verbose: bool) -> Result<()> {
    println!("Planning infrastructure changes...");
    println!("  Playbook: {}", playbook.display());

    // Parse variables
    let vars = parse_variables(variables)?;
    if !vars.is_empty() {
        println!("  Variables:");
        for (k, v) in &vars {
            println!("    {} = {}", k, v);
        }
    }

    // Create playbook runner with plan mode
    let runner = playbook_runner::PlaybookRunner::new(playbook.clone())
        .with_variables(vars.clone())
        .with_verbose(verbose);

    // In plan mode, we execute but don't persist state changes
    // For now, we just show what would be executed
    println!("\nPlan output:");
    println!("{}", "-".repeat(60));

    runner.run()?;

    println!("{}", "-".repeat(60));
    println!("\nPlan complete. No changes were made.");
    println!("Run 'noetl iap apply {}' to apply these changes.", playbook.display());

    Ok(())
}

/// Apply infrastructure changes
async fn iap_apply(playbook: &PathBuf, variables: &[String], auto_approve: bool, verbose: bool) -> Result<()> {
    if !auto_approve {
        println!("This will apply infrastructure changes.");
        print!("Do you want to continue? [y/N] ");
        use std::io::{self, Write};
        io::stdout().flush()?;

        let mut input = String::new();
        io::stdin().read_line(&mut input)?;

        if !input.trim().eq_ignore_ascii_case("y") {
            println!("Apply cancelled.");
            return Ok(());
        }
    }

    println!("Applying infrastructure changes...");
    println!("  Playbook: {}", playbook.display());

    let vars = parse_variables(variables)?;

    let runner = playbook_runner::PlaybookRunner::new(playbook.clone())
        .with_variables(vars)
        .with_verbose(verbose);

    runner.run()?;

    println!("\nApply complete.");
    Ok(())
}

// =============================================================================
// IaP Workspace Management Functions
// =============================================================================

/// Get workspace registry path
fn get_workspace_registry_path() -> PathBuf {
    PathBuf::from(".noetl/workspaces.json")
}

/// Workspace registry entry
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
struct WorkspaceEntry {
    name: String,
    project: String,
    region: String,
    bucket: Option<String>,
    state_path_template: String,
    /// Fully evaluated remote path (computed from template)
    remote_path: Option<String>,
    created_at: String,
    last_used: Option<String>,
}

/// Load workspace registry
fn load_workspace_registry() -> Result<HashMap<String, WorkspaceEntry>> {
    let registry_path = get_workspace_registry_path();
    if registry_path.exists() {
        let content = fs::read_to_string(&registry_path)?;
        Ok(serde_json::from_str(&content).unwrap_or_default())
    } else {
        Ok(HashMap::new())
    }
}

/// Save workspace registry
fn save_workspace_registry(registry: &HashMap<String, WorkspaceEntry>) -> Result<()> {
    let registry_path = get_workspace_registry_path();
    if let Some(parent) = registry_path.parent() {
        fs::create_dir_all(parent)?;
    }
    let content = serde_json::to_string_pretty(registry)?;
    fs::write(&registry_path, content)?;
    Ok(())
}

/// List workspaces
async fn iap_workspace_list(include_remote: bool) -> Result<()> {
    use duckdb::Connection;

    let registry = load_workspace_registry()?;

    // Get current workspace from state
    let state_path = get_state_db_path()?;
    let current_workspace = if state_path.exists() {
        let conn = Connection::open(&state_path)?;
        conn.query_row("SELECT value FROM iap_config WHERE key = 'workspace'", [], |row| {
            row.get::<_, String>(0)
        })
        .ok()
    } else {
        None
    };

    println!("Registered Workspaces:");
    println!("{:<20} {:<20} {:<15} {:<8}", "NAME", "PROJECT", "REGION", "CURRENT");
    println!("{}", "-".repeat(65));

    if registry.is_empty() && current_workspace.is_none() {
        println!("No workspaces registered. Use 'noetl iap init' to create one.");
    } else {
        for (name, entry) in &registry {
            let is_current = current_workspace.as_ref().map(|c| c == name).unwrap_or(false);
            println!(
                "{:<20} {:<20} {:<15} {}",
                name,
                &entry.project,
                &entry.region,
                if is_current { "*" } else { "" }
            );
        }

        // If current workspace isn't in registry, show it anyway
        if let Some(ref current) = current_workspace {
            if !registry.contains_key(current) {
                println!("{:<20} {:<20} {:<15} *  (not in registry)", current, "-", "-");
            }
        }
    }

    if include_remote {
        println!("\nChecking remote workspaces in GCS...");

        // Get bucket from current state or registry
        let bucket = if state_path.exists() {
            let conn = Connection::open(&state_path)?;
            conn.query_row("SELECT value FROM iap_config WHERE key = 'bucket'", [], |row| {
                row.get::<_, String>(0)
            })
            .ok()
        } else {
            None
        };

        if let Some(bucket) = bucket {
            // List workspaces folder in GCS
            let output = std::process::Command::new("gsutil")
                .args(["ls", &format!("gs://{}/workspaces/", bucket)])
                .output();

            match output {
                Ok(out) if out.status.success() => {
                    let stdout = String::from_utf8_lossy(&out.stdout);
                    let remote_workspaces: Vec<&str> = stdout
                        .lines()
                        .filter_map(|line| {
                            line.strip_prefix(&format!("gs://{}/workspaces/", bucket))
                                .and_then(|s| s.strip_suffix("/"))
                        })
                        .collect();

                    if remote_workspaces.is_empty() {
                        println!("No remote workspaces found in gs://{}/workspaces/", bucket);
                    } else {
                        println!("\nRemote Workspaces (gs://{}/workspaces/):", bucket);
                        for ws in remote_workspaces {
                            let is_local = registry.contains_key(ws)
                                || current_workspace.as_ref().map(|c| c == ws).unwrap_or(false);
                            println!("  {} {}", ws, if is_local { "(synced)" } else { "(remote only)" });
                        }
                    }
                }
                Ok(out) => {
                    let stderr = String::from_utf8_lossy(&out.stderr);
                    if !stderr.contains("BucketNotFoundException") {
                        println!("Could not list remote workspaces: {}", stderr);
                    }
                }
                Err(e) => {
                    println!("gsutil not available: {}", e);
                }
            }
        } else {
            println!("No GCS bucket configured. Use 'noetl iap init --bucket <bucket>' first.");
        }
    }

    Ok(())
}

/// Switch to a different workspace
async fn iap_workspace_switch(name: &str, pull_after: bool) -> Result<()> {
    use duckdb::{params, Connection};

    let mut registry = load_workspace_registry()?;

    // Check if workspace exists in registry
    let entry = registry.get(name).cloned();

    if entry.is_none() {
        // Check if workspace exists remotely
        let state_path = get_state_db_path()?;
        if state_path.exists() {
            let conn = Connection::open(&state_path)?;
            let bucket: Option<String> = conn
                .query_row("SELECT value FROM iap_config WHERE key = 'bucket'", [], |row| {
                    row.get(0)
                })
                .ok();

            if let Some(bucket) = bucket {
                let remote_path = format!("gs://{}/workspaces/{}/state.duckdb", bucket, name);
                let check = std::process::Command::new("gsutil")
                    .args(["stat", &remote_path])
                    .output();

                if let Ok(out) = check {
                    if out.status.success() {
                        println!("Workspace '{}' found in remote storage.", name);
                        println!("Creating local entry and pulling state...");

                        // Get project and region from current config
                        let project: String = conn
                            .query_row("SELECT value FROM iap_config WHERE key = 'project'", [], |row| {
                                row.get(0)
                            })
                            .unwrap_or_else(|_| "unknown".to_string());
                        let region: String = conn
                            .query_row("SELECT value FROM iap_config WHERE key = 'region'", [], |row| {
                                row.get(0)
                            })
                            .unwrap_or_else(|_| "us-central1".to_string());

                        // Create registry entry
                        let remote_path = format!("gs://{}/workspaces/{}/state.duckdb", bucket, name);
                        let new_entry = WorkspaceEntry {
                            name: name.to_string(),
                            project: project.clone(),
                            region: region.clone(),
                            bucket: Some(bucket.clone()),
                            state_path_template: "workspaces/{workspace}/state.duckdb".to_string(),
                            remote_path: Some(remote_path),
                            created_at: chrono::Utc::now().to_rfc3339(),
                            last_used: None,
                        };
                        registry.insert(name.to_string(), new_entry);
                        save_workspace_registry(&registry)?;

                        // Update current workspace in state
                        conn.execute(
                            "INSERT OR REPLACE INTO iap_config (key, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP)",
                            params!["workspace", name],
                        )?;

                        // Pull state
                        iap_sync_pull(true).await?;

                        println!("\nSwitched to workspace '{}'.", name);
                        return Ok(());
                    }
                }
            }
        }

        return Err(anyhow::anyhow!(
            "Workspace '{}' not found in registry or remote storage.\n\
             Use 'noetl iap workspace create {}' to create a new workspace, or\n\
             Use 'noetl iap workspace list --remote' to see available remote workspaces.",
            name,
            name
        ));
    }

    // Workspace exists in registry
    let entry = entry.unwrap();
    let state_path = get_state_db_path()?;

    // Update workspace in state database
    let conn = Connection::open(&state_path)?;
    conn.execute(
        "INSERT OR REPLACE INTO iap_config (key, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP)",
        params!["workspace", name],
    )?;
    conn.execute(
        "INSERT OR REPLACE INTO iap_config (key, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP)",
        params!["project", &entry.project],
    )?;
    conn.execute(
        "INSERT OR REPLACE INTO iap_config (key, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP)",
        params!["region", &entry.region],
    )?;
    if let Some(ref bucket) = entry.bucket {
        conn.execute(
            "INSERT OR REPLACE INTO iap_config (key, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP)",
            params!["bucket", bucket],
        )?;
    }
    conn.execute(
        "INSERT OR REPLACE INTO iap_config (key, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP)",
        params!["state_path_template", &entry.state_path_template],
    )?;

    // Update last_used in registry
    if let Some(entry) = registry.get_mut(name) {
        entry.last_used = Some(chrono::Utc::now().to_rfc3339());
    }
    save_workspace_registry(&registry)?;

    println!("Switched to workspace '{}'.", name);
    println!("  Project: {}", entry.project);
    println!("  Region: {}", entry.region);
    if let Some(bucket) = &entry.bucket {
        println!("  Bucket: {}", bucket);
    }

    if pull_after {
        println!("\nPulling state from remote...");
        iap_sync_pull(true).await?;
    }

    Ok(())
}

/// Show current workspace info
async fn iap_workspace_current() -> Result<()> {
    use duckdb::Connection;

    let state_path = get_state_db_path()?;
    if !state_path.exists() {
        return Err(anyhow::anyhow!("No state database found. Run 'noetl iap init' first."));
    }

    let conn = Connection::open(&state_path)?;

    let workspace: String = conn
        .query_row("SELECT value FROM iap_config WHERE key = 'workspace'", [], |row| {
            row.get(0)
        })
        .unwrap_or_else(|_| "default".to_string());

    let project: String = conn
        .query_row("SELECT value FROM iap_config WHERE key = 'project'", [], |row| {
            row.get(0)
        })
        .unwrap_or_else(|_| "-".to_string());

    let region: String = conn
        .query_row("SELECT value FROM iap_config WHERE key = 'region'", [], |row| {
            row.get(0)
        })
        .unwrap_or_else(|_| "-".to_string());

    let bucket: String = conn
        .query_row("SELECT value FROM iap_config WHERE key = 'bucket'", [], |row| {
            row.get(0)
        })
        .unwrap_or_else(|_| "(not configured)".to_string());

    let state_path_template: String = conn
        .query_row(
            "SELECT value FROM iap_config WHERE key = 'state_path_template'",
            [],
            |row| row.get(0),
        )
        .unwrap_or_else(|_| "workspaces/{workspace}/state.duckdb".to_string());

    // Get resource count
    let resource_count: i64 = conn
        .query_row("SELECT COUNT(*) FROM resources", [], |row| row.get(0))
        .unwrap_or(0);

    println!("Current Workspace: {}", workspace);
    println!();
    println!("Configuration:");
    println!("  Project:     {}", project);
    println!("  Region:      {}", region);
    println!("  GCS Bucket:  {}", bucket);
    println!(
        "  State Path:  {}",
        state_path_template.replace("{workspace}", &workspace)
    );
    println!();
    println!("State:");
    println!("  Local DB:    {}", state_path.display());
    println!("  Resources:   {}", resource_count);

    Ok(())
}

/// Create a new workspace
async fn iap_workspace_create(name: &str, from: Option<&str>, switch_to: bool) -> Result<()> {
    use duckdb::Connection;

    let mut registry = load_workspace_registry()?;

    // Check if workspace already exists
    if registry.contains_key(name) {
        return Err(anyhow::anyhow!(
            "Workspace '{}' already exists. Use 'noetl iap workspace switch {}' to switch to it.",
            name,
            name
        ));
    }

    // Get configuration from current state or source workspace
    let (project, region, bucket, state_path_template) = if let Some(source) = from {
        // Clone from existing workspace
        if let Some(entry) = registry.get(source) {
            (
                entry.project.clone(),
                entry.region.clone(),
                entry.bucket.clone(),
                entry.state_path_template.clone(),
            )
        } else {
            return Err(anyhow::anyhow!("Source workspace '{}' not found in registry.", source));
        }
    } else {
        // Get from current state
        let state_path = get_state_db_path()?;
        if !state_path.exists() {
            return Err(anyhow::anyhow!(
                "No state database found. Run 'noetl iap init' first, or use --from to clone from existing workspace."
            ));
        }

        let conn = Connection::open(&state_path)?;
        let project: String = conn
            .query_row("SELECT value FROM iap_config WHERE key = 'project'", [], |row| {
                row.get(0)
            })
            .context("Project not configured. Run 'noetl iap init' first.")?;
        let region: String = conn
            .query_row("SELECT value FROM iap_config WHERE key = 'region'", [], |row| {
                row.get(0)
            })
            .unwrap_or_else(|_| "us-central1".to_string());
        let bucket: Option<String> = conn
            .query_row("SELECT value FROM iap_config WHERE key = 'bucket'", [], |row| {
                row.get(0)
            })
            .ok();
        let state_path_template: String = conn
            .query_row(
                "SELECT value FROM iap_config WHERE key = 'state_path_template'",
                [],
                |row| row.get(0),
            )
            .unwrap_or_else(|_| "workspaces/{workspace}/state.duckdb".to_string());

        (project, region, bucket, state_path_template)
    };

    // Create registry entry
    let remote_path = bucket.as_ref().map(|b| {
        let path = state_path_template.replace("{workspace}", name);
        format!("gs://{}/{}", b, path)
    });
    let entry = WorkspaceEntry {
        name: name.to_string(),
        project: project.clone(),
        region: region.clone(),
        bucket: bucket.clone(),
        state_path_template: state_path_template.clone(),
        remote_path,
        created_at: chrono::Utc::now().to_rfc3339(),
        last_used: None,
    };

    registry.insert(name.to_string(), entry);
    save_workspace_registry(&registry)?;

    println!("Created workspace '{}'.", name);
    println!("  Project: {}", project);
    println!("  Region: {}", region);
    if let Some(ref bucket) = bucket {
        let remote_path = state_path_template.replace("{workspace}", name);
        println!("  Remote: gs://{}/{}", bucket, remote_path);
    }

    if switch_to {
        println!();
        iap_workspace_switch(name, false).await?;
    } else {
        println!("\nTo switch to this workspace, run:");
        println!("  noetl iap workspace switch {}", name);
    }

    Ok(())
}

/// Delete a workspace from registry
async fn iap_workspace_delete(name: &str, delete_remote: bool, force: bool) -> Result<()> {
    use duckdb::Connection;
    use std::io::{self, Write};

    let mut registry = load_workspace_registry()?;

    // Check if it's the current workspace
    let state_path = get_state_db_path()?;
    let current_workspace = if state_path.exists() {
        let conn = Connection::open(&state_path)?;
        conn.query_row("SELECT value FROM iap_config WHERE key = 'workspace'", [], |row| {
            row.get::<_, String>(0)
        })
        .ok()
    } else {
        None
    };

    if current_workspace.as_ref().map(|c| c == name).unwrap_or(false) {
        return Err(anyhow::anyhow!(
            "Cannot delete current workspace '{}'. Switch to a different workspace first.",
            name
        ));
    }

    if !registry.contains_key(name) {
        return Err(anyhow::anyhow!("Workspace '{}' not found in registry.", name));
    }

    if !force {
        print!("Delete workspace '{}'? ", name);
        if delete_remote {
            print!("(including remote state) ");
        }
        print!("[y/N] ");
        io::stdout().flush()?;

        let mut input = String::new();
        io::stdin().read_line(&mut input)?;
        if !input.trim().eq_ignore_ascii_case("y") {
            println!("Aborted.");
            return Ok(());
        }
    }

    // Delete remote state if requested
    if delete_remote {
        if let Some(entry) = registry.get(name) {
            if let Some(ref bucket) = entry.bucket {
                let remote_path = entry.state_path_template.replace("{workspace}", name);
                let gcs_path = format!("gs://{}/{}", bucket, remote_path);

                println!("Deleting remote state: {}", gcs_path);
                let output = std::process::Command::new("gsutil").args(["rm", &gcs_path]).output();

                match output {
                    Ok(out) if out.status.success() => {
                        println!("Remote state deleted.");
                    }
                    Ok(out) => {
                        let stderr = String::from_utf8_lossy(&out.stderr);
                        if !stderr.contains("No URLs matched") {
                            println!("Warning: Could not delete remote state: {}", stderr);
                        }
                    }
                    Err(e) => {
                        println!("Warning: Could not delete remote state: {}", e);
                    }
                }
            }
        }
    }

    // Remove from registry
    registry.remove(name);
    save_workspace_registry(&registry)?;

    println!("Workspace '{}' removed from registry.", name);

    Ok(())
}

/// List resources in state
async fn iap_state_list(resource_type: Option<&str>, format: &str) -> Result<()> {
    use duckdb::Connection;

    let state_path = get_state_db_path()?;
    let conn = Connection::open(&state_path).context("Failed to open state database")?;

    let query = if let Some(rt) = resource_type {
        format!(
            "SELECT resource_id, resource_type, resource_name, status, CAST(updated_at AS VARCHAR) AS updated_at 
             FROM resources WHERE resource_type = '{}' ORDER BY resource_type, resource_name",
            rt
        )
    } else {
        "SELECT resource_id, resource_type, resource_name, status, CAST(updated_at AS VARCHAR) AS updated_at 
         FROM resources ORDER BY resource_type, resource_name"
            .to_string()
    };

    let mut stmt = conn.prepare(&query)?;
    let rows = stmt.query_map([], |row| {
        Ok((
            row.get::<_, String>(0)?,
            row.get::<_, String>(1)?,
            row.get::<_, String>(2)?,
            row.get::<_, String>(3)?,
            row.get::<_, String>(4)?,
        ))
    })?;

    let results: Vec<_> = rows.filter_map(|r| r.ok()).collect();

    if results.is_empty() {
        println!("No resources found in state.");
        return Ok(());
    }

    match format {
        "json" => {
            let json_results: Vec<serde_json::Value> = results
                .iter()
                .map(|(id, rtype, name, status, updated)| {
                    serde_json::json!({
                        "resource_id": id,
                        "resource_type": rtype,
                        "resource_name": name,
                        "status": status,
                        "updated_at": updated
                    })
                })
                .collect();
            println!("{}", serde_json::to_string_pretty(&json_results)?);
        }
        _ => {
            println!(
                "{:<40} {:<15} {:<25} {:<10} {}",
                "RESOURCE ID", "TYPE", "NAME", "STATUS", "UPDATED"
            );
            println!("{}", "-".repeat(100));
            for (id, rtype, name, status, updated) in results {
                println!("{:<40} {:<15} {:<25} {:<10} {}", id, rtype, name, status, updated);
            }
        }
    }

    Ok(())
}

/// Show details for a specific resource
async fn iap_state_show(resource: &str) -> Result<()> {
    use duckdb::Connection;

    let state_path = get_state_db_path()?;
    let conn = Connection::open(&state_path)?;

    let mut stmt = conn.prepare(
        "SELECT resource_id, resource_type, resource_name, provider, project, region, zone,
                status, config, state, created_at, updated_at, last_sync_at, drift_detected
         FROM resources WHERE resource_id = ? OR resource_name = ?",
    )?;

    let result = stmt.query_row([resource, resource], |row| {
        Ok(serde_json::json!({
            "resource_id": row.get::<_, String>(0)?,
            "resource_type": row.get::<_, String>(1)?,
            "resource_name": row.get::<_, String>(2)?,
            "provider": row.get::<_, String>(3)?,
            "project": row.get::<_, Option<String>>(4)?,
            "region": row.get::<_, Option<String>>(5)?,
            "zone": row.get::<_, Option<String>>(6)?,
            "status": row.get::<_, String>(7)?,
            "config": row.get::<_, Option<String>>(8)?,
            "state": row.get::<_, Option<String>>(9)?,
            "created_at": row.get::<_, String>(10)?,
            "updated_at": row.get::<_, String>(11)?,
            "last_sync_at": row.get::<_, Option<String>>(12)?,
            "drift_detected": row.get::<_, bool>(13)?
        }))
    });

    match result {
        Ok(json) => {
            println!("{}", serde_json::to_string_pretty(&json)?);
        }
        Err(_) => {
            println!("Resource not found: {}", resource);
        }
    }

    Ok(())
}

/// Remove a resource from state
async fn iap_state_rm(resource: &str, force: bool) -> Result<()> {
    use duckdb::Connection;

    if !force {
        print!(
            "Remove '{}' from state? This does not destroy the actual resource. [y/N] ",
            resource
        );
        use std::io::{self, Write};
        io::stdout().flush()?;

        let mut input = String::new();
        io::stdin().read_line(&mut input)?;

        if !input.trim().eq_ignore_ascii_case("y") {
            println!("Cancelled.");
            return Ok(());
        }
    }

    let state_path = get_state_db_path()?;
    let conn = Connection::open(&state_path)?;

    let deleted = conn.execute(
        "DELETE FROM resources WHERE resource_id = ? OR resource_name = ?",
        [resource, resource],
    )?;

    if deleted > 0 {
        println!("Resource '{}' removed from state.", resource);
    } else {
        println!("Resource '{}' not found in state.", resource);
    }

    Ok(())
}

/// Execute raw SQL query against state database
async fn iap_state_query(sql: &str) -> Result<()> {
    use std::process::Command;

    let state_path = get_state_db_path()?;

    // Use duckdb CLI for arbitrary queries (the Rust crate has bugs with query_arrow)
    let output = Command::new("duckdb").arg(&state_path).arg("-c").arg(sql).output();

    match output {
        Ok(out) => {
            if out.status.success() {
                print!("{}", String::from_utf8_lossy(&out.stdout));
                if !out.stderr.is_empty() {
                    eprint!("{}", String::from_utf8_lossy(&out.stderr));
                }
                Ok(())
            } else {
                let stderr = String::from_utf8_lossy(&out.stderr);
                Err(anyhow::anyhow!("DuckDB query failed: {}", stderr))
            }
        }
        Err(e) => Err(anyhow::anyhow!(
            "DuckDB CLI not found. Install with: brew install duckdb\nError: {}",
            e
        )),
    }
}

/// Push local state to GCS
async fn iap_sync_push(force: bool) -> Result<()> {
    use duckdb::Connection;

    let state_path = get_state_db_path()?;
    let conn = Connection::open(&state_path)?;

    // Get configuration from state database
    let bucket: String = conn
        .query_row("SELECT value FROM iap_config WHERE key = 'bucket'", [], |row| {
            row.get(0)
        })
        .context("GCS bucket not configured. Run 'noetl iap init --bucket <bucket>' first.")?;

    let workspace: String = conn
        .query_row("SELECT value FROM iap_config WHERE key = 'workspace'", [], |row| {
            row.get(0)
        })
        .unwrap_or_else(|_| "default".to_string());

    let state_path_template: String = conn
        .query_row(
            "SELECT value FROM iap_config WHERE key = 'state_path_template'",
            [],
            |row| row.get(0),
        )
        .unwrap_or_else(|_| "workspaces/{workspace}/state.duckdb".to_string());

    // Build remote path from template
    let remote_path = state_path_template.replace("{workspace}", &workspace);
    let gcs_path = format!("gs://{}/{}", bucket, remote_path);

    if !force {
        print!("Push state to {}? [y/N] ", gcs_path);
        use std::io::{self, Write};
        io::stdout().flush()?;

        let mut input = String::new();
        io::stdin().read_line(&mut input)?;

        if !input.trim().eq_ignore_ascii_case("y") {
            println!("Cancelled.");
            return Ok(());
        }
    }

    println!("Pushing state to {}...", gcs_path);

    // Close connection before copying
    drop(conn);

    // Use gsutil to copy
    let output = std::process::Command::new("gsutil")
        .args(["cp", state_path.to_str().unwrap(), &gcs_path])
        .output()
        .context("Failed to push state to GCS (gsutil not available?)")?;

    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        anyhow::bail!("Failed to push state: {}", stderr);
    }

    println!("State pushed successfully.");
    Ok(())
}

/// Pull state from GCS to local
async fn iap_sync_pull(force: bool) -> Result<()> {
    use duckdb::Connection;

    let state_path = get_state_db_path()?;

    // Try to get config from existing state, or require it
    if !state_path.exists() {
        return Err(anyhow::anyhow!(
            "No local state found. Run 'noetl iap init' first to configure project and bucket."
        ));
    }

    let conn = Connection::open(&state_path)?;

    let bucket: String = conn
        .query_row("SELECT value FROM iap_config WHERE key = 'bucket'", [], |row| {
            row.get(0)
        })
        .context("GCS bucket not configured")?;

    let workspace: String = conn
        .query_row("SELECT value FROM iap_config WHERE key = 'workspace'", [], |row| {
            row.get(0)
        })
        .unwrap_or_else(|_| "default".to_string());

    let state_path_template: String = conn
        .query_row(
            "SELECT value FROM iap_config WHERE key = 'state_path_template'",
            [],
            |row| row.get(0),
        )
        .unwrap_or_else(|_| "workspaces/{workspace}/state.duckdb".to_string());

    // Build remote path from template
    let remote_path = state_path_template.replace("{workspace}", &workspace);
    let gcs_path = format!("gs://{}/{}", bucket, remote_path);

    // Close connection before pulling
    drop(conn);

    if !force {
        print!("Pull state from {}? This will overwrite local state. [y/N] ", gcs_path);
        use std::io::{self, Write};
        io::stdout().flush()?;

        let mut input = String::new();
        io::stdin().read_line(&mut input)?;

        if !input.trim().eq_ignore_ascii_case("y") {
            println!("Cancelled.");
            return Ok(());
        }
    }

    println!("Pulling state from {}...", gcs_path);

    let output = std::process::Command::new("gsutil")
        .args(["cp", &gcs_path, state_path.to_str().unwrap()])
        .output()
        .context("Failed to pull state from GCS (gsutil not available?)")?;

    if !output.status.success() {
        let stderr = String::from_utf8_lossy(&output.stderr);
        anyhow::bail!("Failed to pull state: {}", stderr);
    }

    println!("State pulled successfully.");
    Ok(())
}

/// Show sync status
async fn iap_sync_status() -> Result<()> {
    use duckdb::Connection;

    let state_path = get_state_db_path()?;

    if !state_path.exists() {
        println!("No local state found.");
        return Ok(());
    }

    let conn = Connection::open(&state_path)?;

    let bucket: Result<String, _> = conn.query_row("SELECT value FROM iap_config WHERE key = 'bucket'", [], |row| {
        row.get(0)
    });

    let project: String = conn
        .query_row("SELECT value FROM iap_config WHERE key = 'project'", [], |row| {
            row.get(0)
        })
        .unwrap_or_else(|_| "unknown".to_string());

    let workspace: String = conn
        .query_row("SELECT value FROM iap_config WHERE key = 'workspace'", [], |row| {
            row.get(0)
        })
        .unwrap_or_else(|_| "default".to_string());

    let state_path_template: String = conn
        .query_row(
            "SELECT value FROM iap_config WHERE key = 'state_path_template'",
            [],
            |row| row.get(0),
        )
        .unwrap_or_else(|_| "workspaces/{workspace}/state.duckdb".to_string());

    println!("Local state: {}", state_path.display());
    println!("Project: {}", project);
    println!("Workspace: {}", workspace);

    match bucket {
        Ok(b) => {
            let remote_path = state_path_template.replace("{workspace}", &workspace);
            let gcs_path = format!("gs://{}/{}", b, remote_path);
            println!("Remote: {}", gcs_path);

            // Check if remote exists
            let output = std::process::Command::new("gsutil").args(["ls", &gcs_path]).output();

            match output {
                Ok(o) if o.status.success() => {
                    println!("Remote state: exists");
                }
                _ => {
                    println!("Remote state: not found");
                }
            }
        }
        Err(_) => {
            println!("Remote: not configured");
        }
    }

    // Count local resources
    let count: i64 = conn
        .query_row("SELECT COUNT(*) FROM resources", [], |row| row.get(0))
        .unwrap_or(0);

    println!("Local resources: {}", count);

    Ok(())
}

/// Detect drift between state and actual resources
async fn iap_drift_detect(_resource_type: Option<&str>, _resource: Option<&str>) -> Result<()> {
    println!("Drift detection requires cloud API access.");
    println!("This feature is not yet implemented.");
    println!("\nTo detect drift, the system would:");
    println!("1. Read resources from local state");
    println!("2. Query GCP APIs for current resource state");
    println!("3. Compare and report differences");
    Ok(())
}

/// Show drift report
async fn iap_drift_report(format: &str) -> Result<()> {
    use duckdb::Connection;

    let state_path = get_state_db_path()?;
    let conn = Connection::open(&state_path)?;

    let mut stmt = conn.prepare(
        "SELECT resource_id, detected_at, drift_type, expected_value, actual_value
         FROM drift_log WHERE resolved_at IS NULL ORDER BY detected_at DESC",
    )?;

    let rows = stmt.query_map([], |row| {
        Ok((
            row.get::<_, String>(0)?,
            row.get::<_, String>(1)?,
            row.get::<_, String>(2)?,
            row.get::<_, Option<String>>(3)?,
            row.get::<_, Option<String>>(4)?,
        ))
    })?;

    let results: Vec<_> = rows.filter_map(|r| r.ok()).collect();

    if results.is_empty() {
        println!("No drift detected.");
        return Ok(());
    }

    match format {
        "json" => {
            let json_results: Vec<serde_json::Value> = results
                .iter()
                .map(|(id, detected, dtype, expected, actual)| {
                    serde_json::json!({
                        "resource_id": id,
                        "detected_at": detected,
                        "drift_type": dtype,
                        "expected": expected,
                        "actual": actual
                    })
                })
                .collect();
            println!("{}", serde_json::to_string_pretty(&json_results)?);
        }
        _ => {
            println!("{:<40} {:<20} {:<15}", "RESOURCE", "DETECTED", "DRIFT TYPE");
            println!("{}", "-".repeat(80));
            for (id, detected, dtype, _, _) in results {
                println!("{:<40} {:<20} {:<15}", id, detected, dtype);
            }
        }
    }

    Ok(())
}

/// Parse key=value variables
fn parse_variables(variables: &[String]) -> Result<std::collections::HashMap<String, String>> {
    let mut vars = std::collections::HashMap::new();
    for var in variables {
        let parts: Vec<&str> = var.splitn(2, '=').collect();
        if parts.len() != 2 {
            return Err(anyhow::anyhow!("Invalid variable format: {}. Expected key=value", var));
        }
        vars.insert(parts[0].to_string(), parts[1].to_string());
    }
    Ok(vars)
}

/// Get the default state database path
fn get_state_db_path() -> Result<PathBuf> {
    let path = PathBuf::from(".noetl/state.duckdb");
    Ok(path)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn distributed_payload_accepts_workload_prefixed_overrides() {
        let variables = vec![
            "workload.num_facilities=1".to_string(),
            "workload.patients_per_facility=10".to_string(),
            "workload.enabled=true".to_string(),
        ];

        let payload = build_distributed_payload(None, None, &variables).unwrap();

        assert_eq!(payload["num_facilities"], serde_json::json!(1));
        assert_eq!(payload["patients_per_facility"], serde_json::json!(10));
        assert_eq!(payload["enabled"], serde_json::json!(true));
        assert!(payload.get("workload.num_facilities").is_none());
    }

    #[test]
    fn distributed_payload_keeps_direct_payload_overrides() {
        let variables = vec!["num_facilities=2".to_string()];

        let payload =
            build_distributed_payload(None, Some(r#"{"num_facilities": 1, "page_size": 50}"#), &variables).unwrap();

        assert_eq!(payload["num_facilities"], serde_json::json!(2));
        assert_eq!(payload["page_size"], serde_json::json!(50));
    }

    #[test]
    fn distributed_payload_supports_nested_override_paths() {
        let variables = vec![
            "limits.http.concurrency=8".to_string(),
            "limits.http.window=\"30s\"".to_string(),
        ];

        let payload = build_distributed_payload(None, None, &variables).unwrap();

        assert_eq!(payload["limits"]["http"]["concurrency"], serde_json::json!(8));
        assert_eq!(payload["limits"]["http"]["window"], serde_json::json!("30s"));
    }

    #[test]
    fn auth0_tenant_extracted_from_standard_domain() {
        // ``<tenant>.auth0.com`` shape — the simplest case.
        assert_eq!(
            extract_auth0_tenant_from_domain("acme.auth0.com"),
            Some("acme".to_string())
        );
    }

    #[test]
    fn auth0_tenant_extracted_from_regional_domain() {
        // ``<tenant>.<region>.auth0.com`` — Auth0 regional shape
        // (eu, us, au, jp).  We pull the first label, which is the
        // tenant slug used in the management URL.
        assert_eq!(
            extract_auth0_tenant_from_domain("mestumre-development.us.auth0.com"),
            Some("mestumre-development".to_string())
        );
        assert_eq!(
            extract_auth0_tenant_from_domain("acme.eu.auth0.com"),
            Some("acme".to_string())
        );
    }

    #[test]
    fn auth0_tenant_returns_none_for_custom_domain() {
        // CNAMEd custom domains can't be parsed for a tenant slug.
        // Return None so the caller falls back to ``<tenant>`` in
        // the dashboard hint.
        assert_eq!(extract_auth0_tenant_from_domain("login.example.com"), None);
        assert_eq!(extract_auth0_tenant_from_domain(""), None);
    }

    #[test]
    fn auth0_tenant_normalises_whitespace_and_case() {
        assert_eq!(
            extract_auth0_tenant_from_domain("  ACME.AUTH0.COM  "),
            Some("acme".to_string())
        );
        assert_eq!(
            extract_auth0_tenant_from_domain("acme.auth0.com/"),
            Some("acme".to_string())
        );
    }

    #[test]
    fn auth0_tenant_and_region_extracts_both_pieces() {
        assert_eq!(
            extract_auth0_tenant_and_region("acme.auth0.com"),
            Some(("acme".to_string(), None))
        );
        assert_eq!(
            extract_auth0_tenant_and_region("mestumre-development.us.auth0.com"),
            Some(("mestumre-development".to_string(), Some("us".to_string())))
        );
        assert_eq!(
            extract_auth0_tenant_and_region("acme.eu.auth0.com"),
            Some(("acme".to_string(), Some("eu".to_string())))
        );
        assert_eq!(
            extract_auth0_tenant_and_region("acme.au.auth0.com"),
            Some(("acme".to_string(), Some("au".to_string())))
        );
        assert_eq!(
            extract_auth0_tenant_and_region("acme.jp.auth0.com"),
            Some(("acme".to_string(), Some("jp".to_string())))
        );
    }

    #[test]
    fn auth0_tenant_and_region_rejects_unknown_region_label() {
        // Defensive: an unrecognised middle label is more likely a
        // misconfigured domain than a new Auth0 region we just
        // haven't heard of.  Returning None forces the dashboard-URL
        // builder to fall back to the ``<tenant>`` placeholder
        // rather than constructing a URL Auth0 won't recognise.
        assert_eq!(
            extract_auth0_tenant_and_region("acme.xx.auth0.com"),
            None
        );
        assert_eq!(
            extract_auth0_tenant_and_region("a.b.c.auth0.com"),
            None
        );
    }

    #[test]
    fn auth0_tenant_and_region_handles_non_auth0_domains() {
        assert_eq!(extract_auth0_tenant_and_region("login.example.com"), None);
        assert_eq!(extract_auth0_tenant_and_region(""), None);
    }

    #[test]
    fn build_auth0_dashboard_url_default_region() {
        // ``<tenant>.auth0.com`` — no region label in domain ⇒ no
        // region in dashboard URL.
        assert_eq!(
            build_auth0_dashboard_url("acme.auth0.com", "Abc123"),
            Some(
                "https://manage.auth0.com/dashboard/acme/applications/Abc123/settings"
                    .to_string()
            )
        );
    }

    #[test]
    fn build_auth0_dashboard_url_regional_tenant() {
        // Regression for noetl/ai-meta#18 — the previously printed
        // URL was ``dashboard/mestumre-development/applications/...``
        // (missing the ``us`` region segment) and 404'd in the
        // browser.  Auth0's actual format is
        // ``dashboard/<region>/<tenant>/applications/...``.
        assert_eq!(
            build_auth0_dashboard_url(
                "mestumre-development.us.auth0.com",
                "Jqop7YoaiZalLHdBRo5ScNQ1RJhbhbDN"
            ),
            Some(
                "https://manage.auth0.com/dashboard/us/mestumre-development/applications/Jqop7YoaiZalLHdBRo5ScNQ1RJhbhbDN/settings"
                    .to_string()
            )
        );
        assert_eq!(
            build_auth0_dashboard_url("acme.eu.auth0.com", "EuClientId"),
            Some(
                "https://manage.auth0.com/dashboard/eu/acme/applications/EuClientId/settings"
                    .to_string()
            )
        );
    }

    #[test]
    fn build_auth0_dashboard_url_returns_none_for_custom_domain() {
        // CNAMEd custom domains can't be parsed for tenant/region.
        // Caller falls back to a ``<tenant>`` placeholder URL.
        assert_eq!(
            build_auth0_dashboard_url("login.example.com", "anything"),
            None
        );
    }

    #[test]
    fn build_auth0_dashboard_url_trims_whitespace_in_client_id() {
        assert_eq!(
            build_auth0_dashboard_url("acme.auth0.com", "  Abc123  "),
            Some(
                "https://manage.auth0.com/dashboard/acme/applications/Abc123/settings"
                    .to_string()
            )
        );
    }

    #[test]
    fn gateway_auth_expired_exit_code_is_77() {
        // The exit code is part of the CLI's public contract — shell
        // scripts can branch on it (``[ $? -eq 77 ] && noetl auth
        // login``).  Don't change it without a deliberate plan.
        assert_eq!(GATEWAY_AUTH_EXPIRED_EXIT_CODE, 77);
    }

    #[test]
    fn local_port_parses_from_loopback_server_urls() {
        // Standard kubectl-port-forward target shapes.  IPv4 only —
        // kubectl always binds to the IPv4 loopback even on dual-stack
        // hosts, so the IPv6 case isn't relevant in practice.
        assert_eq!(parse_local_port_from_server_url("http://127.0.0.1:18082"), Some(18082));
        assert_eq!(parse_local_port_from_server_url("http://localhost:8082"), Some(8082));
    }

    #[test]
    fn local_port_rejects_non_loopback_urls() {
        // Non-loopback URLs don't make sense as port-forward targets —
        // the kubectl tunnel always lands on localhost.
        assert_eq!(parse_local_port_from_server_url("https://gateway.mestumre.dev"), None);
        assert_eq!(parse_local_port_from_server_url("http://example.com:8082"), None);
        assert_eq!(parse_local_port_from_server_url("http://10.0.0.5:8082"), None);
    }

    #[test]
    fn local_port_returns_none_for_missing_port() {
        // URLs without an explicit port are also invalid — kubectl
        // needs a concrete port to bind locally.
        assert_eq!(parse_local_port_from_server_url("http://localhost"), None);
        assert_eq!(parse_local_port_from_server_url("not-a-url"), None);
    }

    #[test]
    fn port_forward_target_requires_kube_fields() {
        // Without ``kube_context`` and ``kube_namespace``, the helper
        // returns None so the caller can print the configuration hint.
        let mut ctx = Context::new("http://127.0.0.1:18082".to_string());
        assert!(port_forward_target(&ctx).is_none());

        ctx.kube_context = Some("gke_my-project_us-central1_my-cluster".to_string());
        assert!(port_forward_target(&ctx).is_none(), "namespace still missing");

        ctx.kube_namespace = Some("noetl".to_string());
        let target = port_forward_target(&ctx).expect("all required fields set");
        assert_eq!(target.0, "gke_my-project_us-central1_my-cluster");
        assert_eq!(target.1, "noetl");
        // Defaults applied for optional fields.
        assert_eq!(target.2, "noetl"); // service
        assert_eq!(target.3, 18082); // local port from server_url
        assert_eq!(target.4, 8082); // remote port default
    }

    #[test]
    fn port_forward_target_honours_explicit_service_and_remote_port() {
        let mut ctx = Context::new("http://localhost:30090".to_string());
        ctx.kube_context = Some("kind-noetl".to_string());
        ctx.kube_namespace = Some("gateway".to_string());
        ctx.kube_service = Some("noetl-gateway".to_string());
        ctx.kube_remote_port = Some(8090);
        let (kctx, ns, svc, local, remote) = port_forward_target(&ctx).unwrap();
        assert_eq!(kctx, "kind-noetl");
        assert_eq!(ns, "gateway");
        assert_eq!(svc, "noetl-gateway");
        assert_eq!(local, 30090);
        assert_eq!(remote, 8090);
    }

    #[test]
    fn port_forward_pid_path_is_under_dot_noetl() {
        // The PID file path is part of the operational surface — `noetl
        // context port-forward --stop` reads it.  Pin the layout so a
        // refactor doesn't silently move it and leave stale daemons.
        let path = port_forward_pid_path("gke-pf").expect("home dir resolves");
        let p = path.to_string_lossy();
        assert!(p.ends_with(".noetl/port-forwards/gke-pf.pid"), "got {}", p);
    }

    #[test]
    fn auth0_parses_from_runtime_contract() {
        // Full auth0 block — every field populated.
        let contract = serde_json::json!({
            "gateway_version": "2.10.0",
            "auth0": {
                "domain": "acme.auth0.com",
                "client_id": "abc123",
                "redirect_uri": "https://app.example.com/login",
                "audience": "https://api.example.com"
            }
        });
        let (domain, client_id, redirect_uri, audience) =
            parse_auth0_block_from_contract(&contract).expect("auth0 present");
        assert_eq!(domain, "acme.auth0.com");
        assert_eq!(client_id, "abc123");
        assert_eq!(redirect_uri, "https://app.example.com/login");
        assert_eq!(audience, "https://api.example.com");
    }

    #[test]
    fn auth0_parses_partial_block_with_only_domain_and_client_id() {
        // Audience often omitted (this tenant uses ID tokens, not
        // audience-scoped access tokens).  Missing fields ⇒ empty
        // strings; caller decides whether to skip or clear.
        let contract = serde_json::json!({
            "auth0": {
                "domain": "mestumre-development.us.auth0.com",
                "client_id": "Jqop7YoaiZalLHdBRo5ScNQ1RJhbhbDN"
            }
        });
        let (domain, client_id, redirect_uri, audience) =
            parse_auth0_block_from_contract(&contract).unwrap();
        assert_eq!(domain, "mestumre-development.us.auth0.com");
        assert_eq!(client_id, "Jqop7YoaiZalLHdBRo5ScNQ1RJhbhbDN");
        assert_eq!(redirect_uri, "");
        assert_eq!(audience, "");
    }

    #[test]
    fn auth0_missing_or_empty_block_returns_none() {
        // Older gateway image (no auth0 block at all).
        let contract = serde_json::json!({"gateway_version":"2.9.0"});
        assert!(parse_auth0_block_from_contract(&contract).is_none());

        // Empty auth0 object.
        let contract = serde_json::json!({"auth0":{}});
        assert!(parse_auth0_block_from_contract(&contract).is_none());

        // Auth0 block with explicitly-empty domain — treated as absent.
        let contract = serde_json::json!({"auth0":{"domain":"   "}});
        assert!(parse_auth0_block_from_contract(&contract).is_none());

        // Auth0 block where ``auth0`` is the wrong type.
        let contract = serde_json::json!({"auth0":"acme.auth0.com"});
        assert!(parse_auth0_block_from_contract(&contract).is_none());
    }

    #[test]
    fn gateway_url_normalisation_strips_trailing_slash_and_whitespace() {
        assert_eq!(normalise_gateway_url("https://gateway.mestumre.dev"), "https://gateway.mestumre.dev");
        assert_eq!(normalise_gateway_url("https://gateway.mestumre.dev/"), "https://gateway.mestumre.dev");
        assert_eq!(normalise_gateway_url("  https://gateway.mestumre.dev/  "), "https://gateway.mestumre.dev");
        assert_eq!(normalise_gateway_url("http://127.0.0.1:18082/"), "http://127.0.0.1:18082");
        assert_eq!(normalise_gateway_url(""), "");
    }

    #[test]
    fn effective_context_returns_named_context_when_override_set() {
        // ``--context <name>`` resolves against config.contexts, returning
        // the named context regardless of which one is current.
        let mut config = Config::default();
        let mut prod_ctx = Context::new("https://gateway.prod.example.com".to_string());
        prod_ctx.runtime = "distributed".to_string();
        let mut pf_ctx = Context::new("http://127.0.0.1:18082".to_string());
        pf_ctx.runtime = "auto".to_string();
        config.contexts.insert("gke-prod".to_string(), prod_ctx);
        config.contexts.insert("gke-pf".to_string(), pf_ctx);
        config.current_context = Some("gke-prod".to_string());

        // Without --context: current_context wins.
        let (name, ctx) = effective_context(None, &config).unwrap();
        assert_eq!(name, "gke-prod");
        assert_eq!(ctx.server_url, "https://gateway.prod.example.com");

        // With --context=gke-pf: override wins.
        let (name, ctx) = effective_context(Some("gke-pf"), &config).unwrap();
        assert_eq!(name, "gke-pf");
        assert_eq!(ctx.server_url, "http://127.0.0.1:18082");
    }

    #[test]
    fn effective_context_falls_back_to_current_when_override_is_none() {
        let mut config = Config::default();
        config
            .contexts
            .insert("only".to_string(), Context::new("http://localhost:8082".to_string()));
        config.current_context = Some("only".to_string());

        let (name, _ctx) = effective_context(None, &config).unwrap();
        assert_eq!(name, "only");
    }

    #[test]
    fn effective_context_returns_none_when_no_current_and_no_override() {
        let config = Config::default();
        assert!(effective_context(None, &config).is_none());
    }
}
