cdk/cdk_install.py

#!/usr/bin/env python3
"""
Interactive CDK installer for llm_topic_modeller.

Walks through demo vs production deployment profiles, writes config/cdk_config.env
and cdk.json, optionally runs cdk deploy, post-deploy Cognito callback fixups (API,
no second deploy), and post_cdk_build_quickstart.py.

Usage examples::

    # Full interactive install + deploy + quickstart
    python cdk_install.py

    # Demo sandbox, config only
    python cdk_install.py --profile demo --config-only --yes

    # Production with flags (partial non-interactive)
    python cdk_install.py --profile production --vpc-name my-vpc \\
        --cert-arn arn:aws:acm:eu-west-2:123:certificate/abc \\
        --domain llm-topic.example.com --yes

    # Redeploy using existing config, skip quickstart
    python cdk_install.py --deploy-only --skip-quickstart

    # Remove existing stacks before a clean install (non-interactive)
    python cdk_install.py --profile headless --vpc-name my-vpc \\
        --force-delete-stacks --yes

    # Headless batch (S3 → Lambda → one-shot ECS direct mode)
    python cdk_install.py --profile headless --vpc-name my-vpc --yes
    python cdk_install.py --profile production --headless --vpc-name my-vpc --yes

    # Optional scheduled DynamoDB usage log export to S3
    python cdk_install.py --profile demo --vpc-name my-vpc --yes \\
        --dynamo-usage-log-export --dynamo-export-schedule-time 06:00 \\
        --dynamo-export-schedule-days weekdays

    # Headless with S3 output email alerts + IAM reader user
    python cdk_install.py --profile headless --vpc-name my-vpc --yes \\
        --headless-output-notifications \\
        --headless-notify-email analyst@example.com

    # Enable Bedrock model invocation logging (S3 log bucket + CloudWatch)
    python cdk_install.py --profile headless --vpc-name my-vpc --yes \\
        --bedrock-model-invocation-logging

    # Prefer Resource Groups + tags; leave AppRegistry off (default)
    python cdk_install.py --profile production --vpc-name my-vpc --yes \\
        --application-resource-group --application-name my-prefix-llm-topic-modeller

    # Opt in to legacy AppRegistry / myApplications (existing customers only)
    python cdk_install.py --profile demo --vpc-name my-vpc --yes --appregistry
"""

from __future__ import annotations

import argparse
import ipaddress
import json
import os
import re
import secrets
import shutil
import subprocess
import sys
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Sequence, Tuple

CDK_DIR = Path(__file__).resolve().parent
REPO_ROOT = CDK_DIR.parent
CONFIG_DIR = CDK_DIR / "config"
ENV_PATH = CONFIG_DIR / "cdk_config.env"
APP_CONFIG_ENV_PATH = CONFIG_DIR / "app_config.env"
APP_CONFIG_ENV_EXAMPLE = CONFIG_DIR / "app_config.env.example"
PI_AGENT_ENV_PATH = CONFIG_DIR / "pi_agent.env"
PI_AGENT_ENV_EXAMPLE = REPO_ROOT / "config" / "pi_agent.env.example"
PI_ALB_ROUTING_MODES = ("path", "host", "both")
CDK_JSON_PATH = CDK_DIR / "cdk.json"
CDK_JSON_EXAMPLE = CDK_DIR / "cdk.json.example"
QUICKSTART_SCRIPT = CDK_DIR / "post_cdk_build_quickstart.py"

REGIONAL_STACK = "SummarisationStack"
CLOUDFRONT_STACK = "SummarisationStackCloudfront"
CLOUDFRONT_STACK_REGION = "us-east-1"
APPREGISTRY_STACK_SUFFIX = "AppRegistryStack"

DEMO_PRESET: Dict[str, str] = {
    "USE_ECS_EXPRESS_MODE": "True",
    "ECS_EXPRESS_USE_PUBLIC_SUBNETS": "True",
    "ENABLE_ECS_VPC_INTERFACE_ENDPOINTS": "True",
    "USE_CLOUDFRONT": "False",
    "RUN_USEAST_STACK": "False",
    "ENABLE_RESOURCE_DELETE_PROTECTION": "False",
    "ENABLE_APPREGISTRY": "False",
    "ENABLE_APPLICATION_RESOURCE_GROUP": "True",
    "ACM_SSL_CERTIFICATE_ARN": "",
    "SSL_CERTIFICATE_DOMAIN": "",
}

PRODUCTION_PRESET: Dict[str, str] = {
    "USE_ECS_EXPRESS_MODE": "False",
    "USE_CLOUDFRONT": "True",
    "RUN_USEAST_STACK": "True",
    "ENABLE_RESOURCE_DELETE_PROTECTION": "True",
    "ENABLE_APPREGISTRY": "False",
    "ENABLE_APPLICATION_RESOURCE_GROUP": "True",
}

HEADLESS_PRESET: Dict[str, str] = {
    "USE_ECS_EXPRESS_MODE": "False",
    "ECS_EXPRESS_USE_PUBLIC_SUBNETS": "True",
    "USE_CLOUDFRONT": "False",
    "RUN_USEAST_STACK": "False",
    "ENABLE_RESOURCE_DELETE_PROTECTION": "False",
    "ENABLE_APPREGISTRY": "False",
    "ENABLE_APPLICATION_RESOURCE_GROUP": "True",
    "ENABLE_HEADLESS_DEPLOYMENT": "True",
    "ENABLE_S3_BATCH_ECS_TRIGGER": "True",
    "COGNITO_AUTH": "False",
    "ACM_SSL_CERTIFICATE_ARN": "",
    "SSL_CERTIFICATE_DOMAIN": "",
}

DEFAULT_CDK_JSON_CONTEXT: Dict[str, Any] = {
    "@aws-cdk/aws-ec2:restrictDefaultSecurityGroup": False,
}


# ---------------------------------------------------------------------------
# Wizard helpers
# ---------------------------------------------------------------------------


def ask(prompt: str, default: str = "") -> str:
    suffix = f" [{default}]" if default else ""
    value = input(f"{prompt}{suffix}: ").strip()
    return value or default


def ask_yes_no(prompt: str, default: bool = True) -> bool:
    default_label = "Y/n" if default else "y/N"
    while True:
        raw = input(f"{prompt} ({default_label}): ").strip().lower()
        if not raw:
            return default
        if raw in ("y", "yes"):
            return True
        if raw in ("n", "no"):
            return False
        print("Please enter y or n.")


def ask_choice(prompt: str, options: Sequence[str], default_index: int = 0) -> int:
    print(prompt)
    for i, opt in enumerate(options):
        marker = "*" if i == default_index else " "
        print(f"  {marker} {i + 1}) {opt}")
    while True:
        raw = input(
            f"Choice [1-{len(options)}] (default {default_index + 1}): "
        ).strip()
        if not raw:
            return default_index
        try:
            idx = int(raw) - 1
            if 0 <= idx < len(options):
                return idx
        except ValueError:
            pass
        print("Invalid choice.")


def timestamp() -> str:
    return datetime.now().strftime("%Y%m%d-%H%M%S")


def backup_file(path: Path) -> Optional[Path]:
    if not path.is_file():
        return None
    backup = path.with_suffix(path.suffix + f".bak.{timestamp()}")
    shutil.copy2(path, backup)
    print(f"Backed up {path.name} -> {backup.name}")
    return backup


# ---------------------------------------------------------------------------
# Python / cdk.json
# ---------------------------------------------------------------------------


def _venv_python_paths() -> List[Path]:
    candidates: List[Path] = []
    virtual_env = os.environ.get("VIRTUAL_ENV", "").strip()
    if virtual_env:
        root = Path(virtual_env)
        if os.name == "nt":
            candidates.append(root / "Scripts" / "python.exe")
        else:
            candidates.append(root / "bin" / "python")

    rel_venvs = [
        CDK_DIR / ".venv",
        REPO_ROOT / ".venv",
        REPO_ROOT.parent / ".venv",
    ]
    for venv_root in rel_venvs:
        if os.name == "nt":
            candidates.append(venv_root / "Scripts" / "python.exe")
        else:
            candidates.append(venv_root / "bin" / "python")

    for name in ("python3", "python"):
        found = shutil.which(name)
        if found:
            candidates.append(Path(found))

    seen: set[str] = set()
    unique: List[Path] = []
    for p in [Path(sys.executable)] + candidates:
        key = str(p.resolve()) if p.exists() else str(p)
        if key not in seen:
            seen.add(key)
            unique.append(p)
    return unique


def _resolve_node_executable() -> Optional[str]:
    return shutil.which("node")


def _jsii_import_failure_hint(stderr: str) -> str:
    if "JSONDecodeError" not in stderr and "Expecting value" not in stderr:
        return ""
    return (
        "\nThis usually means the JSII Node.js helper process failed to start.\n"
        "On the deployment machine, check:\n"
        "  1. Node.js is installed and on PATH: node --version (LTS 18–22 recommended)\n"
        "  2. Reinstall CDK Python deps: pip install --force-reinstall -r requirements.txt\n"
        "  3. Test manually: python -c \"from aws_cdk import App; print('ok')\"\n"
        "  4. If still failing, run: set JSII_DEBUG=1  (then retry synth for details)"
    )


def verify_node_for_jsii() -> str:
    """Return the node executable path, or exit with guidance if missing."""
    node = _resolve_node_executable()
    if not node:
        raise SystemExit(
            "Node.js not found on PATH. aws-cdk-lib (JSII) requires Node.js.\n"
            "Install Node.js LTS from https://nodejs.org/ and ensure 'node' is on PATH."
        )
    try:
        result = subprocess.run(
            [node, "--version"],
            capture_output=True,
            text=True,
            timeout=10,
            check=False,
        )
    except (OSError, subprocess.TimeoutExpired) as exc:
        raise SystemExit(f"Could not run Node.js ({node}): {exc}") from exc
    if result.returncode != 0:
        raise SystemExit(
            f"Node.js ({node}) returned exit code {result.returncode}.\n"
            f"{(result.stderr or result.stdout or '').strip()}"
        )
    version = (result.stdout or result.stderr or "").strip()
    print(f"Node.js for JSII: {node} ({version})")
    return node


def _python_has_aws_cdk(python_exe: Path) -> Tuple[bool, str]:
    if not python_exe.is_file():
        return False, "not found"
    cmd = [
        str(python_exe),
        "-c",
        "import aws_cdk; print(getattr(aws_cdk, '__version__', 'unknown'))",
    ]
    try:
        result = subprocess.run(
            cmd,
            cwd=str(CDK_DIR),
            capture_output=True,
            text=True,
            timeout=30,
            check=False,
        )
        if result.returncode == 0:
            version = (result.stdout or "").strip().splitlines()[-1]
            return True, version
        detail = (result.stderr or result.stdout or "import failed").strip()
        hint = _jsii_import_failure_hint(detail)
        if hint:
            detail = f"{detail[:400]}{hint}"
        return False, detail[:800]
    except (OSError, subprocess.TimeoutExpired) as exc:
        return False, str(exc)


def discover_python_candidates() -> List[Tuple[Path, str]]:
    found: List[Tuple[Path, str]] = []
    for candidate in _venv_python_paths():
        ok, detail = _python_has_aws_cdk(candidate)
        if ok:
            found.append((candidate.resolve(), detail))
    return found


def resolve_python_executable(
    override: Optional[str] = None,
    interactive: bool = True,
    assume_yes: bool = False,
) -> Path:
    if override:
        path = Path(override).resolve()
        ok, detail = _python_has_aws_cdk(path)
        if not ok:
            raise SystemExit(
                f"Python at {path} cannot import aws-cdk-lib: {detail}\n"
                "Install CDK deps: pip install -r requirements.txt"
            )
        print(f"Using Python: {path} (aws-cdk-lib {detail})")
        return path

    candidates = discover_python_candidates()
    if not candidates:
        raise SystemExit(
            "No Python interpreter with aws-cdk-lib found.\n"
            "Activate your venv and run: pip install -r requirements.txt\n"
            "Or pass --python /path/to/python"
        )

    if len(candidates) == 1 or assume_yes or not interactive:
        path, version = candidates[0]
        print(f"Using Python: {path} (aws-cdk-lib {version})")
        return path

    print("Python interpreters with aws-cdk-lib:")
    labels = [f"{p} (aws-cdk-lib {v})" for p, v in candidates]
    idx = ask_choice("Select Python for cdk.json", labels, default_index=0)
    path, version = candidates[idx]
    print(f"Selected: {path} (aws-cdk-lib {version})")
    return path


def format_cdk_app_command(python_exe: Path) -> str:
    """Format the cdk.json ``app`` command for the current OS."""
    py = str(python_exe)
    if os.name == "nt":
        py = py.replace("/", "\\")
    return f"{py} app.py"


def load_cdk_json_context(existing_path: Path) -> Dict[str, Any]:
    if existing_path.is_file():
        try:
            data = json.loads(existing_path.read_text(encoding="utf-8"))
            ctx = data.get("context")
            if isinstance(ctx, dict):
                return dict(ctx)
        except json.JSONDecodeError:
            pass
    if CDK_JSON_EXAMPLE.is_file():
        try:
            data = json.loads(CDK_JSON_EXAMPLE.read_text(encoding="utf-8"))
            ctx = data.get("context")
            if isinstance(ctx, dict):
                return dict(ctx)
        except json.JSONDecodeError:
            pass
    return dict(DEFAULT_CDK_JSON_CONTEXT)


def write_cdk_json(
    python_exe: Path,
    *,
    force: bool = False,
    skip: bool = False,
) -> Optional[Path]:
    if skip:
        if not CDK_JSON_PATH.is_file():
            raise SystemExit(
                f"{CDK_JSON_PATH.name} not found; cannot use --skip-cdk-json."
            )
        print(f"Keeping existing {CDK_JSON_PATH.name} (--skip-cdk-json)")
        return CDK_JSON_PATH

    if CDK_JSON_PATH.is_file() and force:
        backup_file(CDK_JSON_PATH)
    elif not CDK_JSON_PATH.is_file():
        pass
    else:
        # Update app only — backup first
        backup_file(CDK_JSON_PATH)

    context = load_cdk_json_context(CDK_JSON_PATH)
    payload = {
        "app": format_cdk_app_command(python_exe),
        "output": "cdk.out",
        "context": context,
    }
    CDK_JSON_PATH.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
    print(f"Wrote {CDK_JSON_PATH}")
    return CDK_JSON_PATH


def build_cdk_subprocess_env(
    values: Optional[Dict[str, str]] = None,
) -> Dict[str, str]:
    """
    Environment for child Python / CDK processes.

    Merges config/cdk_config.env over the current process environment so stale
    empty defaults from an earlier cdk_config import during the wizard do not
    block load_dotenv (override=False) in app.py.
    """
    env = os.environ.copy()
    env["CDK_CONFIG_PATH"] = str(ENV_PATH)
    cdk_folder = ""
    if values:
        cdk_folder = (values.get("CDK_FOLDER") or "").strip()
    if not cdk_folder and ENV_PATH.is_file():
        cdk_folder = (read_env_file(ENV_PATH).get("CDK_FOLDER") or "").strip()
    if not cdk_folder:
        cdk_folder = str(CDK_DIR).replace("\\", "/") + "/"
    env["CDK_FOLDER"] = cdk_folder

    file_values: Dict[str, str] = {}
    if ENV_PATH.is_file():
        file_values = read_env_file(ENV_PATH)
    elif values:
        file_values = dict(values)

    for key, val in file_values.items():
        env[key] = str(val)
    return env


def apply_cdk_runtime_env(values: Dict[str, str]) -> None:
    """Expose written config to app.py imports (smoke test, cdk synth/deploy)."""
    os.environ.update(build_cdk_subprocess_env(values))


def smoke_test_python_app(python_exe: Path) -> None:
    verify_node_for_jsii()
    cmd = [str(python_exe), "-c", "import aws_cdk; import app"]
    print("Smoke test: import aws_cdk and app ...")
    result = subprocess.run(
        cmd,
        cwd=str(CDK_DIR),
        env=build_cdk_subprocess_env(),
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        output = (result.stderr or result.stdout or "").strip()
        print(output)
        hint = _jsii_import_failure_hint(output)
        if hint:
            print(hint)
        raise SystemExit(
            "Python smoke test failed. Fix CDK dependencies on this machine before deploying."
        )


def run_smoke_test_if_needed(
    python_exe: Optional[Path],
    args: argparse.Namespace,
) -> None:
    if args.config_only or args.skip_cdk_json or not python_exe:
        return
    smoke_test_python_app(python_exe)


# ---------------------------------------------------------------------------
# AWS discovery
# ---------------------------------------------------------------------------


def get_aws_identity(region: Optional[str] = None) -> Tuple[str, str]:
    import boto3

    session = boto3.Session(region_name=region)
    sts = session.client("sts")
    ident = sts.get_caller_identity()
    account = ident["Account"]
    resolved_region = (
        region
        or session.region_name
        or os.environ.get("AWS_REGION")
        or os.environ.get("AWS_DEFAULT_REGION")
        or "eu-west-2"
    )
    return account, resolved_region


def resolve_cdk_executable() -> str:
    """Return the AWS CDK CLI on PATH (cdk.cmd on Windows)."""
    for name in ("cdk", "cdk.cmd", "cdk.exe"):
        found = shutil.which(name)
        if found:
            return found
    raise SystemExit(
        "AWS CDK CLI not found on PATH.\n"
        "Install: npm install -g aws-cdk  (or use npx aws-cdk)"
    )


def check_cdk_cli() -> str:
    cdk_exe = resolve_cdk_executable()
    result = subprocess.run(
        [cdk_exe, "--version"],
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        raise SystemExit(
            "AWS CDK CLI not found on PATH.\n"
            "Install: npm install -g aws-cdk  (or use npx aws-cdk)"
        )
    version = (result.stdout or result.stderr or "").strip()
    if version.startswith("Python "):
        raise SystemExit(
            "CDK CLI check failed: 'cdk' resolved to the Python interpreter instead "
            "of the AWS CDK CLI. Install aws-cdk globally (npm install -g aws-cdk) "
            "and ensure it is on PATH before python.exe."
        )
    return version


def cdk_bootstrap_needed(account: str, region: str) -> bool:
    import boto3
    from botocore.exceptions import ClientError

    cfn = boto3.client("cloudformation", region_name=region)
    try:
        cfn.describe_stacks(StackName="CDKToolkit")
        return False
    except ClientError as exc:
        if "does not exist" in str(exc):
            return True
        raise


def list_vpcs(region: str) -> List[Dict[str, str]]:
    import boto3

    ec2 = boto3.client("ec2", region_name=region)
    response = ec2.describe_vpcs()
    vpcs: List[Dict[str, str]] = []
    for vpc in response.get("Vpcs", []):
        name = ""
        for tag in vpc.get("Tags", []):
            if tag.get("Key") == "Name":
                name = tag.get("Value", "")
                break
        cidrs = vpc_cidr_blocks_from_describe(vpc)
        vpcs.append(
            {
                "id": vpc["VpcId"],
                "name": name or vpc["VpcId"],
                "cidr": cidrs[0] if cidrs else "",
                "cidrs": ",".join(cidrs),
            }
        )
    return sorted(vpcs, key=lambda x: x["name"])


def vpc_cidr_blocks_from_describe(vpc: Dict[str, Any]) -> List[str]:
    """Primary and associated IPv4 CIDR blocks for one ``describe_vpcs`` entry."""
    blocks: List[str] = []
    seen: set[str] = set()
    primary = (vpc.get("CidrBlock") or "").strip()
    if primary and primary not in seen:
        seen.add(primary)
        blocks.append(primary)
    for assoc in vpc.get("CidrBlockAssociationSet") or []:
        cidr = (assoc.get("CidrBlock") or "").strip()
        if cidr and cidr not in seen:
            seen.add(cidr)
            blocks.append(cidr)
    return blocks


def list_vpc_cidr_blocks_in_region(region: str) -> List[str]:
    """All IPv4 VPC CIDR blocks in the account/region (primary + associations)."""
    import boto3

    ec2 = boto3.client("ec2", region_name=region)
    blocks: List[str] = []
    seen: set[str] = set()
    for vpc in ec2.describe_vpcs().get("Vpcs", []):
        for cidr in vpc_cidr_blocks_from_describe(vpc):
            if cidr not in seen:
                seen.add(cidr)
                blocks.append(cidr)
    return blocks


VPC_CIDR_PREFIX_LEN = 24
SUBNET_CIDR_PREFIX_LEN = 28
# ECS Express provisions a managed ALB in each public subnet (8+ free IPs required).
# /28 subnets (~11 usable IPs) exhaust quickly with VPC interface endpoints and tasks.
EXPRESS_PUBLIC_SUBNET_CIDR_PREFIX_LEN = 27
PUBLIC_SUBNET_CIDR_PREFIX_LEN = 26
PRIVATE_SUBNET_CIDR_PREFIX_LEN = 28
EXPRESS_ALB_MIN_SUBNET_PREFIX_LEN = 27
_VPC_CIDR_SEARCH_SUPERNETS = ("10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16")


def parse_ipv4_networks(cidrs: Sequence[str]) -> List[ipaddress.IPv4Network]:
    networks: List[ipaddress.IPv4Network] = []
    for cidr in cidrs:
        raw = (cidr or "").strip()
        if not raw:
            continue
        try:
            networks.append(ipaddress.ip_network(raw, strict=False))
        except ValueError:
            continue
    return networks


def suggest_vpc_cidr_block(
    existing_vpc_cidrs: Sequence[str],
    *,
    prefix_len: int = VPC_CIDR_PREFIX_LEN,
    search_supernets: Sequence[str] = _VPC_CIDR_SEARCH_SUPERNETS,
) -> str:
    """Return the lowest non-overlapping ``prefix_len`` block in RFC1918 space."""
    occupied = parse_ipv4_networks(existing_vpc_cidrs)
    for supernet_cidr in search_supernets:
        supernet = ipaddress.ip_network(supernet_cidr, strict=False)
        for candidate in supernet.subnets(new_prefix=prefix_len):
            if any(candidate.overlaps(block) for block in occupied):
                continue
            return str(candidate)
    raise ValueError(
        f"No unused /{prefix_len} VPC CIDR found in region "
        f"({len(occupied)} existing block(s))."
    )


def validate_new_vpc_cidr_format(cidr: str) -> Optional[str]:
    """Return an error when ``cidr`` is not a valid AWS VPC IPv4 block."""
    raw = (cidr or "").strip()
    if not raw:
        return "NEW_VPC_CIDR is required when creating a new VPC."
    try:
        network = ipaddress.ip_network(raw, strict=False)
    except ValueError:
        return f"NEW_VPC_CIDR={raw!r} is not a valid IPv4 CIDR block."
    if network.version != 4:
        return f"NEW_VPC_CIDR must be IPv4 (got {raw!r})."
    if network.prefixlen < 16 or network.prefixlen > 28:
        return (
            f"NEW_VPC_CIDR={raw!r} must use a prefix length between /16 and /28 "
            f"(got /{network.prefixlen})."
        )
    if not network.is_private:
        return (
            f"NEW_VPC_CIDR={raw!r} should be in RFC1918 private space "
            "(10.0.0.0/8, 172.16.0.0/12, or 192.168.0.0/16)."
        )
    return None


def validate_new_vpc_cidr_no_overlap(
    cidr: str,
    existing_vpc_cidrs: Sequence[str],
) -> Optional[str]:
    """Return an error when ``cidr`` overlaps an existing VPC block in the region."""
    format_error = validate_new_vpc_cidr_format(cidr)
    if format_error:
        return format_error
    candidate = ipaddress.ip_network(cidr.strip(), strict=False)
    for block in parse_ipv4_networks(existing_vpc_cidrs):
        if candidate.overlaps(block):
            return (
                f"NEW_VPC_CIDR={candidate!s} overlaps existing VPC CIDR {block!s} "
                "in this account/region."
            )
    return None


def validate_new_vpc_cidr(
    cidr: str,
    existing_vpc_cidrs: Sequence[str],
) -> Optional[str]:
    """Validate format and regional non-overlap for a new VPC CIDR."""
    return validate_new_vpc_cidr_no_overlap(cidr, existing_vpc_cidrs)


def canonicalize_vpc_cidr(cidr: str) -> str:
    """Normalize a VPC CIDR string (e.g. 10.0.0.0/24)."""
    return str(ipaddress.ip_network(cidr.strip(), strict=False))


def subnet_cidr_prefix_len_for_tier(answers: "InstallAnswers", tier: str) -> int:
    """Return installer subnet sizing for public/private tiers and deployment mode."""
    if tier == "public":
        if answers_use_public_subnets_only(answers):
            return EXPRESS_PUBLIC_SUBNET_CIDR_PREFIX_LEN
        return PUBLIC_SUBNET_CIDR_PREFIX_LEN
    return PRIVATE_SUBNET_CIDR_PREFIX_LEN


def validate_public_subnet_cidr_for_express(cidr: str) -> Optional[str]:
    """Return an error when a public subnet is too small for Express managed ALB."""
    raw = (cidr or "").strip()
    if not raw:
        return None
    try:
        network = ipaddress.ip_network(raw, strict=False)
    except ValueError:
        return f"Public subnet CIDR {raw!r} is not valid."
    if network.prefixlen > EXPRESS_ALB_MIN_SUBNET_PREFIX_LEN:
        return (
            f"Public subnet {network!s} is too small for ECS Express Mode: the managed "
            f"ALB needs at least 8 free IP addresses per subnet. Use /"
            f"{EXPRESS_ALB_MIN_SUBNET_PREFIX_LEN} or larger (e.g. /27 in a /24 VPC)."
        )
    return None


def validate_new_vpc_cidr_env_values(values: Dict[str, str]) -> List[str]:
    """Live regional overlap check for NEW_VPC_CIDR before deploy."""
    new_cidr = (values.get("NEW_VPC_CIDR") or "").strip()
    if not new_cidr or (values.get("VPC_NAME") or "").strip():
        return []
    region = (values.get("AWS_REGION") or "").strip()
    if not region:
        return ["AWS_REGION is required to validate NEW_VPC_CIDR."]
    occupied = list_vpc_cidr_blocks_in_region(region)
    err = validate_new_vpc_cidr(new_cidr, occupied)
    return [err] if err else []


def prompt_new_vpc_cidr(answers: "InstallAnswers", *, interactive: bool) -> None:
    """Fill ``answers.new_vpc_cidr`` via auto-suggest and/or manual wizard prompt."""
    region = answers.aws_region
    occupied = list_vpc_cidr_blocks_in_region(region)

    if answers.new_vpc_cidr:
        err = validate_new_vpc_cidr(answers.new_vpc_cidr, occupied)
        if err:
            raise SystemExit(err)
        answers.new_vpc_cidr = canonicalize_vpc_cidr(answers.new_vpc_cidr)
        return

    auto_prompt = (
        f"Auto-select lowest available /{VPC_CIDR_PREFIX_LEN} VPC CIDR " f"in {region}?"
    )

    def _select_suggested() -> str:
        suggested = suggest_vpc_cidr_block(occupied)
        print(f"Selected VPC CIDR: {suggested}")
        return suggested

    def _prompt_manual(default_cidr: str) -> str:
        while True:
            raw = ask("New VPC CIDR", default_cidr)
            err = validate_new_vpc_cidr(raw, occupied)
            if err:
                print(err)
                continue
            return canonicalize_vpc_cidr(raw)

    if interactive:
        try:
            default_cidr = suggest_vpc_cidr_block(occupied)
            can_auto_select = True
        except ValueError as exc:
            print(f"Warning: {exc}")
            default_cidr = ""
            can_auto_select = False

        if can_auto_select and ask_yes_no(auto_prompt, default=True):
            answers.new_vpc_cidr = _select_suggested()
        else:
            if not default_cidr:
                print(
                    f"Enter a non-overlapping RFC1918 VPC CIDR (/16–/28) for {region}."
                )
            answers.new_vpc_cidr = _prompt_manual(default_cidr)
        return

    try:
        answers.new_vpc_cidr = _select_suggested()
    except ValueError as exc:
        raise SystemExit(
            f"Could not auto-select a VPC CIDR in {region}. "
            f"Pass --new-vpc-cidr with a non-overlapping RFC1918 /16–/28 block. {exc}"
        ) from exc


def suggest_subnet_cidr_blocks(
    vpc_cidr: str,
    existing_subnet_cidrs: Sequence[str],
    count: int,
    *,
    prefix_len: int = SUBNET_CIDR_PREFIX_LEN,
    reserved_cidrs: Optional[Sequence[str]] = None,
) -> List[str]:
    """Return ``count`` lowest non-overlapping ``prefix_len`` blocks inside ``vpc_cidr``."""
    if count < 1:
        return []
    vpc_net = ipaddress.ip_network(vpc_cidr.strip(), strict=False)
    occupied = parse_ipv4_networks(existing_subnet_cidrs)
    if reserved_cidrs:
        occupied.extend(parse_ipv4_networks(reserved_cidrs))

    suggestions: List[str] = []
    for candidate in vpc_net.subnets(new_prefix=prefix_len):
        if any(candidate.overlaps(block) for block in occupied):
            continue
        suggestions.append(str(candidate))
        occupied.append(candidate)
        if len(suggestions) >= count:
            return suggestions

    raise ValueError(
        f"Only {len(suggestions)} available /{prefix_len} subnet block(s) in "
        f"{vpc_cidr}; need {count}."
    )


def list_availability_zones(region: str) -> List[str]:
    import boto3

    ec2 = boto3.client("ec2", region_name=region)
    zones = ec2.describe_availability_zones(
        Filters=[{"Name": "state", "Values": ["available"]}]
    )
    return [z["ZoneName"] for z in zones.get("AvailabilityZones", [])]


def list_subnets_in_vpc(vpc_id: str, region: str) -> List[Dict[str, str]]:
    sys.path.insert(0, str(CDK_DIR))
    from cdk_functions import _get_existing_subnets_in_vpc

    data = _get_existing_subnets_in_vpc(vpc_id)
    subnets: List[Dict[str, str]] = []
    for name, info in data.get("by_name", {}).items():
        subnets.append(
            {
                "name": name,
                "id": info.get("id", ""),
                "cidr": info.get("cidr", ""),
                "az": info.get("az", ""),
            }
        )
    return sorted(subnets, key=lambda x: x["name"])


def list_acm_certificates(region: str) -> List[Dict[str, str]]:
    import boto3

    acm = boto3.client("acm", region_name=region)
    certs: List[Dict[str, str]] = []
    paginator = acm.get_paginator("list_certificates")
    for page in paginator.paginate(CertificateStatuses=["ISSUED"]):
        for summary in page.get("CertificateSummaryList", []):
            arn = summary["CertificateArn"]
            detail = acm.describe_certificate(CertificateArn=arn)["Certificate"]
            domain = detail.get("DomainName", "")
            sans = detail.get("SubjectAlternativeNames", [])
            label = domain
            if sans and sans[0] != domain:
                label = f"{domain} (+{len(sans) - 1} SANs)"
            certs.append({"arn": arn, "domain": domain, "label": label})
    return certs


def list_igws_for_vpc(vpc_id: str, region: str) -> List[Dict[str, str]]:
    import boto3

    ec2 = boto3.client("ec2", region_name=region)
    response = ec2.describe_internet_gateways(
        Filters=[{"Name": "attachment.vpc-id", "Values": [vpc_id]}]
    )
    return [
        {"id": igw["InternetGatewayId"]} for igw in response.get("InternetGateways", [])
    ]


def list_albs_in_vpc(vpc_id: str, region: str) -> List[Dict[str, str]]:
    import boto3

    elbv2 = boto3.client("elbv2", region_name=region)
    response = elbv2.describe_load_balancers()
    albs: List[Dict[str, str]] = []
    for lb in response.get("LoadBalancers", []):
        if lb.get("VpcId") == vpc_id:
            albs.append(
                {
                    "arn": lb["LoadBalancerArn"],
                    "dns": lb["DNSName"],
                    "name": lb["LoadBalancerName"],
                }
            )
    return albs


# ---------------------------------------------------------------------------
# Env builder / writer
# ---------------------------------------------------------------------------


def format_list_env(values: Sequence[str], *, use_single_quotes: bool = False) -> str:
    if not values:
        return "[]"
    if use_single_quotes:
        inner = ", ".join(f"'{v}'" for v in values)
    else:
        inner = ", ".join(f'"{v}"' for v in values)
    return f"[{inner}]"


SUBNET_TIER_MODES = ("auto", "existing", "create")


def parse_subnet_name_csv(raw: str) -> List[str]:
    return [part.strip() for part in raw.split(",") if part.strip()]


def resolve_subnet_tier_modes(args: argparse.Namespace) -> Tuple[str, str]:
    """CLI: --subnet-mode sets both tiers; per-tier flags override individually."""
    default = getattr(args, "subnet_mode", None) or "auto"
    public = getattr(args, "public_subnet_mode", None) or default
    private = getattr(args, "private_subnet_mode", None) or default
    return public, private


def apply_subnet_cli_flags(args: argparse.Namespace, answers: "InstallAnswers") -> None:
    if getattr(args, "public_subnet_names", None):
        answers.public_subnet_names = parse_subnet_name_csv(args.public_subnet_names)
    if getattr(args, "private_subnet_names", None):
        answers.private_subnet_names = parse_subnet_name_csv(args.private_subnet_names)


def lookup_subnets_by_name(
    vpc_subnets: Sequence[Dict[str, str]],
) -> Dict[str, Dict[str, str]]:
    return {subnet["name"]: subnet for subnet in vpc_subnets if subnet.get("name")}


def fill_existing_subnet_tier_metadata(
    names: Sequence[str],
    lookup: Dict[str, Dict[str, str]],
    *,
    cidrs_out: List[str],
    azs_out: List[str],
) -> List[str]:
    """Resolve CIDR/AZ for existing subnet names (order preserved)."""
    errors: List[str] = []
    cidrs_out.clear()
    azs_out.clear()
    for name in names:
        info = lookup.get(name)
        if not info:
            errors.append(f"Subnet '{name}' was not found in the VPC.")
            continue
        cidr = (info.get("cidr") or "").strip()
        az = (info.get("az") or "").strip()
        if not az:
            errors.append(f"Subnet '{name}' has no availability zone in AWS.")
            continue
        cidrs_out.append(cidr)
        azs_out.append(az)
    return errors


def enrich_existing_subnet_details_from_aws(answers: "InstallAnswers") -> List[str]:
    """Look up CIDR/AZ for tiers using existing named subnets (wizard or CLI)."""
    if answers.vpc_mode != "existing" or not answers.vpc_name:
        return []

    tiers: List[Tuple[str, str, List[str], List[str], List[str]]] = []
    if answers.public_subnet_mode == "existing" and answers.public_subnet_names:
        tiers.append(
            (
                "Public",
                answers.public_subnet_mode,
                answers.public_subnet_names,
                answers.public_subnet_cidrs,
                answers.public_subnet_azs,
            )
        )
    if answers.private_subnet_mode == "existing" and answers.private_subnet_names:
        tiers.append(
            (
                "Private",
                answers.private_subnet_mode,
                answers.private_subnet_names,
                answers.private_subnet_cidrs,
                answers.private_subnet_azs,
            )
        )
    if not tiers:
        return []

    vpc_list = list_vpcs(answers.aws_region)
    vpc_id = next(
        (vpc["id"] for vpc in vpc_list if vpc["name"] == answers.vpc_name),
        None,
    )
    if not vpc_id:
        return [f"VPC '{answers.vpc_name}' not found in {answers.aws_region}."]

    lookup = lookup_subnets_by_name(list_subnets_in_vpc(vpc_id, answers.aws_region))
    errors: List[str] = []
    for _label, _mode, names, cidrs_out, azs_out in tiers:
        if len(cidrs_out) == len(names) == len(azs_out) and names:
            continue
        errors.extend(
            fill_existing_subnet_tier_metadata(
                names,
                lookup,
                cidrs_out=cidrs_out,
                azs_out=azs_out,
            )
        )
    return errors


def apply_subnet_tier_env(
    values: Dict[str, str],
    answers: "InstallAnswers",
    *,
    tier: str,
    mode: str,
) -> None:
    """Write PUBLIC_* or PRIVATE_* subnet keys for one tier (auto | existing | create)."""
    prefix = "PUBLIC" if tier == "public" else "PRIVATE"
    names = (
        answers.public_subnet_names
        if tier == "public"
        else answers.private_subnet_names
    )
    cidrs = (
        answers.public_subnet_cidrs
        if tier == "public"
        else answers.private_subnet_cidrs
    )
    azs = answers.public_subnet_azs if tier == "public" else answers.private_subnet_azs

    if mode == "auto":
        values[f"{prefix}_SUBNETS_TO_USE"] = ""
        values[f"{prefix}_SUBNET_CIDR_BLOCKS"] = ""
        values[f"{prefix}_SUBNET_AVAILABILITY_ZONES"] = ""
    elif mode == "existing":
        values[f"{prefix}_SUBNETS_TO_USE"] = format_list_env(names)
        if names and len(cidrs) == len(names) == len(azs):
            values[f"{prefix}_SUBNET_CIDR_BLOCKS"] = format_list_env(
                cidrs, use_single_quotes=True
            )
            values[f"{prefix}_SUBNET_AVAILABILITY_ZONES"] = format_list_env(
                azs, use_single_quotes=True
            )
        else:
            values[f"{prefix}_SUBNET_CIDR_BLOCKS"] = ""
            values[f"{prefix}_SUBNET_AVAILABILITY_ZONES"] = ""
    else:
        values[f"{prefix}_SUBNETS_TO_USE"] = format_list_env(names)
        values[f"{prefix}_SUBNET_CIDR_BLOCKS"] = format_list_env(
            cidrs, use_single_quotes=True
        )
        values[f"{prefix}_SUBNET_AVAILABILITY_ZONES"] = format_list_env(
            azs, use_single_quotes=True
        )


def validate_subnet_answers(answers: "InstallAnswers") -> List[str]:
    errors: List[str] = []
    if answers.vpc_mode != "existing":
        return errors
    tiers: List[tuple] = [
        (
            "Public",
            answers.public_subnet_mode,
            answers.public_subnet_names,
            answers.public_subnet_cidrs,
        ),
    ]
    if not answers_use_express_mode(answers) and not answers_use_public_subnets_only(
        answers
    ):
        tiers.append(
            (
                "Private",
                answers.private_subnet_mode,
                answers.private_subnet_names,
                answers.private_subnet_cidrs,
            )
        )
    for label, mode, names, cidrs in tiers:
        if mode == "existing" and not names:
            errors.append(
                f"{label} subnets: provide at least one existing subnet name."
            )
        if mode == "create" and not names:
            errors.append(f"{label} subnets: create mode requires subnet names.")
        if mode == "create" and names and cidrs and len(names) != len(cidrs):
            errors.append(f"{label} subnets: CIDR count must match subnet name count.")
        if label == "Public" and answers_use_express_mode(answers) and cidrs:
            for cidr in cidrs:
                subnet_error = validate_public_subnet_cidr_for_express(cidr)
                if subnet_error:
                    errors.append(subnet_error)
    return errors


def ask_subnet_tier_mode(tier_label: str) -> str:
    idx = ask_choice(
        f"{tier_label} subnets",
        [
            "Auto-discover suitable subnets",
            "Use existing named subnets",
            "Create new stack-specific subnets (name + CIDR + AZ)",
        ],
        default_index=0,
    )
    return SUBNET_TIER_MODES[idx]


def configure_subnet_tier(
    answers: "InstallAnswers",
    tier: str,
    mode: str,
    vpc_subnets: List[Dict[str, str]],
    azs: Sequence[str],
    *,
    interactive: bool,
    vpc_cidr: str = "",
    reserved_subnet_cidrs: Optional[Sequence[str]] = None,
) -> None:
    """Collect subnet names/CIDRs for one tier in the wizard."""
    is_public = tier == "public"
    label = "Public" if is_public else "Private"
    prefix_label = "Public" if is_public else "Private"

    if mode == "auto":
        return

    if mode == "existing":
        names = (
            answers.public_subnet_names if is_public else answers.private_subnet_names
        )
        if interactive and not names:
            print(
                f"Enter {label.lower()} subnet names (comma-separated), or pick from:"
            )
            for subnet in vpc_subnets:
                print(f"  - {subnet['name']} ({subnet['cidr']}, {subnet['az']})")
            names = parse_subnet_name_csv(ask(f"{label} subnet names", ""))
        if is_public:
            answers.public_subnet_names = names
            cidrs_out = answers.public_subnet_cidrs
            azs_out = answers.public_subnet_azs
        else:
            answers.private_subnet_names = names
            cidrs_out = answers.private_subnet_cidrs
            azs_out = answers.private_subnet_azs
        if names:
            lookup = lookup_subnets_by_name(vpc_subnets)
            missing = fill_existing_subnet_tier_metadata(
                names,
                lookup,
                cidrs_out=cidrs_out,
                azs_out=azs_out,
            )
            for err in missing:
                print(f"Warning: {err}")
        return

    n_az = len(azs)
    prefix_label = "Public" if tier == "public" else "Private"
    prefix_len = subnet_cidr_prefix_len_for_tier(answers, tier)
    names = [f"{answers.cdk_prefix}{prefix_label}Subnet{i + 1}" for i in range(n_az)]
    if is_public:
        answers.public_subnet_names = names
        cidrs_list = answers.public_subnet_cidrs
        azs_list = answers.public_subnet_azs
        cidr_base = 0
    else:
        answers.private_subnet_names = names
        cidrs_list = answers.private_subnet_cidrs
        azs_list = answers.private_subnet_azs
        cidr_base = 10

    if interactive:
        print(f"Suggested AZs for {label.lower()} subnets: {', '.join(azs)}")
        auto_assign = ask_yes_no(
            f"Auto-assign lowest available /{prefix_len} CIDR blocks "
            f"for {label.lower()} subnets?",
            default=True,
        )
    else:
        auto_assign = bool(vpc_cidr)

    existing_subnet_cidrs = [
        subnet.get("cidr", "") for subnet in vpc_subnets if subnet.get("cidr")
    ]
    suggested: Optional[List[str]] = None
    if auto_assign and vpc_cidr:
        try:
            suggested = suggest_subnet_cidr_blocks(
                vpc_cidr,
                existing_subnet_cidrs,
                len(names),
                prefix_len=prefix_len,
                reserved_cidrs=reserved_subnet_cidrs,
            )
        except ValueError as exc:
            print(f"Warning: {exc}")
            if not interactive:
                raise SystemExit(str(exc)) from exc

    cidrs_list.clear()
    azs_list.clear()
    for i, name in enumerate(names):
        if suggested:
            cidr = suggested[i]
            if interactive:
                print(f"  {name}: {cidr}")
        elif interactive:
            default_cidr = f"10.0.{cidr_base + i}.0/{prefix_len}"
            if vpc_cidr:
                try:
                    default_cidr = suggest_subnet_cidr_blocks(
                        vpc_cidr,
                        existing_subnet_cidrs + cidrs_list,
                        1,
                        prefix_len=prefix_len,
                        reserved_cidrs=reserved_subnet_cidrs,
                    )[0]
                except ValueError:
                    pass
            cidr = ask(f"CIDR for {label.lower()} {name}", default_cidr)
        else:
            cidr = f"10.0.{cidr_base + i}.0/{prefix_len}"
        cidrs_list.append(cidr)
        azs_list.append(azs[i % len(azs)])


@dataclass
class InstallAnswers:
    profile: str = "demo"
    aws_account_id: str = ""
    aws_region: str = ""
    cdk_prefix: str = ""
    cognito_domain_prefix: str = ""
    s3_log_bucket_name: str = ""
    s3_output_bucket_name: str = ""
    github_branch: str = "main"
    vpc_mode: str = "existing"  # new | existing
    vpc_name: str = ""
    new_vpc_cidr: str = ""
    public_subnet_mode: str = "auto"  # auto | existing | create
    private_subnet_mode: str = "auto"  # auto | existing | create
    public_subnet_names: List[str] = field(default_factory=list)
    private_subnet_names: List[str] = field(default_factory=list)
    public_subnet_cidrs: List[str] = field(default_factory=list)
    private_subnet_cidrs: List[str] = field(default_factory=list)
    public_subnet_azs: List[str] = field(default_factory=list)
    private_subnet_azs: List[str] = field(default_factory=list)
    existing_igw_id: str = ""
    existing_alb_arn: str = ""
    existing_alb_dns: str = ""
    acm_cert_arn: str = ""
    ssl_domain: str = ""
    cloudfront_geo: str = ""
    enable_pi_express: bool = False
    enable_pi_legacy: bool = False
    enable_service_connect: bool = False
    enable_s3_batch: bool = False
    enable_headless: bool = False
    enable_headless_output_notifications: bool = False
    headless_output_notify_email: str = ""
    headless_output_iam_user_name: str = ""
    enable_dynamo_usage_log_export: bool = False
    dynamo_export_schedule_time: str = "06:00"
    dynamo_export_schedule_days: str = "daily"
    dynamo_export_s3_key: str = ""
    dynamo_export_date_attribute: str = "timestamp"
    enable_bedrock_model_invocation_logging: bool = False
    bedrock_invocation_s3_prefix: str = "bedrock-logs"
    bedrock_invocation_log_group_name: str = ""
    bedrock_invocation_log_retention_days: str = "90"
    enable_appregistry: bool = False
    enable_application_resource_group: bool = True
    application_name: str = ""
    application_tag_key: str = "Application"
    ecs_memory: str = "8192"
    pi_alb_routing: str = "path"
    pi_alb_path_prefix: str = "/agent"
    pi_alb_host_header: str = ""
    pi_alb_listener_rule_priority: str = ""
    pi_gradio_port: str = "7862"
    sc_discovery_name: str = "llm-topic"
    pi_default_provider: str = "amazon-bedrock"
    write_pi_agent_env: bool = True
    overwrite_pi_agent_env: bool = False
    write_app_config_env: bool = True
    overwrite_app_config_env: bool = False
    custom_overrides: Dict[str, str] = field(default_factory=dict)
    python_path: Optional[str] = None

    @property
    def pi_enabled(self) -> bool:
        return self.enable_pi_express or self.enable_pi_legacy


# Cognito hosted-UI prefix domains cannot contain these substrings (AWS docs).
COGNITO_DOMAIN_RESERVED_SUBSTRINGS = ("amazon", "cognito", "aws")


def sanitize_cognito_domain_prefix(raw: str, *, max_length: int = 63) -> str:
    """Normalize a Cognito domain prefix: allowed chars, no reserved words."""
    prefix = re.sub(r"[^a-z0-9-]", "-", (raw or "").lower())
    for reserved in COGNITO_DOMAIN_RESERVED_SUBSTRINGS:
        prefix = prefix.replace(reserved, "")
    prefix = re.sub(r"-{2,}", "-", prefix).strip("-")
    if max_length > 0:
        prefix = prefix[:max_length].strip("-")
    if not prefix or not re.match(r"^[a-z0-9]", prefix):
        return "llm-topic-app"
    prefix = prefix.rstrip("-")
    if not prefix:
        return "llm-topic-app"
    for reserved in COGNITO_DOMAIN_RESERVED_SUBSTRINGS:
        if reserved in prefix:
            return "llm-topic-app"
    return prefix


def default_cognito_domain_prefix_from_cdk_prefix(cdk_prefix: str) -> str:
    """Derive a short installer default from CDK_PREFIX (globally unique hint)."""
    slug = re.sub(r"[^a-z0-9-]", "-", (cdk_prefix or "").lower()).strip("-")
    return sanitize_cognito_domain_prefix(slug, max_length=20)


def validate_cognito_domain_prefix(prefix: str) -> Optional[str]:
    """Return an error message when prefix violates Cognito domain rules."""
    cleaned = (prefix or "").strip().lower()
    if not cleaned:
        return "COGNITO_USER_POOL_DOMAIN_PREFIX is required."
    reserved = [word for word in COGNITO_DOMAIN_RESERVED_SUBSTRINGS if word in cleaned]
    if reserved:
        return (
            "COGNITO_USER_POOL_DOMAIN_PREFIX cannot contain reserved words: "
            + ", ".join(reserved)
            + "."
        )
    if not re.match(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$", cleaned):
        return (
            "COGNITO_USER_POOL_DOMAIN_PREFIX must use lowercase letters, "
            "numbers, and hyphens only."
        )
    return None


def normalize_pi_path_prefix(raw: str) -> str:
    segment = (raw or "agent").strip().strip("/")
    return f"/{segment}" if segment else "/agent"


def default_pi_listener_priority(use_cloudfront: bool = False) -> str:
    """Default Pi path/host rule priority (1–2 reserved for CloudFront / Express rules)."""
    del use_cloudfront  # kept for call-site compatibility
    return "3"


def derive_ecs_resource_names(cdk_prefix: str) -> Dict[str, str]:
    """ECS/CodeBuild resource names from CDK_PREFIX (matches cdk_config.py defaults)."""
    prefix = (cdk_prefix or "").strip()
    if not prefix:
        return {}
    return {
        "CLUSTER_NAME": f"{prefix}Cluster",
        "ECS_SERVICE_NAME": f"{prefix}ECSService",
        "ECS_EXPRESS_SERVICE_NAME": f"{prefix}ECSService",
        "ECS_PI_EXPRESS_SERVICE_NAME": f"{prefix}PiExpressService",
        "ECS_PI_SERVICE_NAME": f"{prefix}PiAgentService",
        "COGNITO_USER_POOL_CLIENT_SECRET_NAME": f"{prefix}ParamCognitoSecret",
        "CODEBUILD_PROJECT_NAME": f"{prefix}CodeBuildProject",
        "CODEBUILD_PI_PROJECT_NAME": f"{prefix}CodeBuildPiProject",
    }


def derive_s3_bucket_names(cdk_prefix: str) -> Dict[str, str]:
    """S3 bucket names from CDK_PREFIX (must be written to cdk_config.env explicitly)."""
    prefix = (cdk_prefix or "").strip().lower()
    if not prefix:
        return {}
    return {
        "S3_LOG_CONFIG_BUCKET_NAME": f"{prefix}s3-logs",
        "S3_OUTPUT_BUCKET_NAME": f"{prefix}s3-output",
    }


S3_BUCKET_NAME_MAX_LEN = 63


def normalize_s3_bucket_name(raw: str) -> str:
    """Lowercase S3 bucket name with allowed characters only (3–63 chars)."""
    name = re.sub(r"[^a-z0-9.-]", "-", (raw or "").lower())
    name = re.sub(r"-{2,}", "-", name).strip("-")
    if len(name) < 3:
        name = f"{name}-bucket".strip("-")
    return name[:S3_BUCKET_NAME_MAX_LEN]


def suggest_available_s3_bucket_name(
    preferred: str,
    account_id: str,
    *,
    s3_client: Any = None,
    max_attempts: int = 8,
) -> str:
    """Return a globally available S3 bucket name, preferring ``preferred``."""
    from cdk_functions import resolve_s3_bucket_availability

    account = re.sub(r"[^0-9]", "", account_id or "")
    bases = [preferred]
    if account:
        bases.extend(
            [
                f"{account}-{preferred}",
                f"{preferred}-{account}",
            ]
        )
    seen: set[str] = set()
    for base in bases:
        candidate = normalize_s3_bucket_name(base)
        if not candidate or candidate in seen:
            continue
        seen.add(candidate)
        status, _ = resolve_s3_bucket_availability(candidate, s3_client=s3_client)
        if status == "available":
            return candidate
    for _ in range(max_attempts):
        suffix = secrets.token_hex(3)
        candidate = normalize_s3_bucket_name(f"{account}-{preferred}-{suffix}")
        if candidate in seen:
            continue
        seen.add(candidate)
        status, _ = resolve_s3_bucket_availability(candidate, s3_client=s3_client)
        if status == "available":
            return candidate
    raise RuntimeError(
        f"Could not find an available S3 bucket name near {preferred!r}. "
        "Set S3_LOG_CONFIG_BUCKET_NAME / S3_OUTPUT_BUCKET_NAME manually."
    )


def suggest_available_cognito_domain_prefix(
    preferred: str,
    account_id: str,
    region: str,
    *,
    cognito_client: Any = None,
    max_attempts: int = 8,
) -> str:
    """Return an available Cognito hosted UI domain prefix in ``region``."""
    from cdk_functions import resolve_cognito_domain_prefix_availability

    account = re.sub(r"[^0-9]", "", account_id or "")
    preferred_clean = sanitize_cognito_domain_prefix(preferred, max_length=63)
    candidates = [preferred_clean]
    if account:
        candidates.extend(
            [
                sanitize_cognito_domain_prefix(
                    f"{account}-{preferred_clean}", max_length=63
                ),
                sanitize_cognito_domain_prefix(
                    f"{preferred_clean}-{account[-8:]}", max_length=63
                ),
            ]
        )
    seen: set[str] = set()
    for candidate in candidates:
        if not candidate or candidate in seen:
            continue
        seen.add(candidate)
        if (
            resolve_cognito_domain_prefix_availability(
                candidate, region_name=region, cognito_client=cognito_client
            )
            == "available"
        ):
            return candidate
    for _ in range(max_attempts):
        suffix = secrets.token_hex(2)
        candidate = sanitize_cognito_domain_prefix(
            f"{account}-{preferred_clean}-{suffix}", max_length=63
        )
        if not candidate or candidate in seen:
            continue
        seen.add(candidate)
        if (
            resolve_cognito_domain_prefix_availability(
                candidate, region_name=region, cognito_client=cognito_client
            )
            == "available"
        ):
            return candidate
    raise RuntimeError(
        f"Could not find an available Cognito domain prefix near {preferred!r}. "
        "Set COGNITO_USER_POOL_DOMAIN_PREFIX manually."
    )


def _prompt_globally_unique_s3_bucket(
    label: str,
    preferred: str,
    account_id: str,
    *,
    interactive: bool,
    assume_yes: bool,
    cli_override: str = "",
    s3_client: Any = None,
) -> str:
    from cdk_functions import resolve_s3_bucket_availability

    if cli_override:
        name = normalize_s3_bucket_name(cli_override)
        status, _ = resolve_s3_bucket_availability(name, s3_client=s3_client)
        if status == "globally_taken":
            raise SystemExit(
                f"{label}: bucket name {name!r} is taken globally by another AWS "
                "account. Choose a different --s3-log-bucket / --s3-output-bucket."
            )
        return name

    preferred = normalize_s3_bucket_name(preferred)
    status, _ = resolve_s3_bucket_availability(preferred, s3_client=s3_client)
    if status == "owned":
        print(f"{label}: using existing bucket in this account: {preferred}")
        return preferred
    if status == "available":
        print(f"{label}: bucket name available: {preferred}")
        return preferred

    suggested = suggest_available_s3_bucket_name(
        preferred, account_id, s3_client=s3_client
    )
    print(
        f"{label}: S3 bucket name {preferred!r} is taken globally by another AWS "
        f"account (S3 names are unique worldwide)."
    )
    if not interactive:
        if assume_yes:
            print(f"{label}: using suggested name: {suggested}")
            return suggested
        raise SystemExit(
            f"{label}: bucket name {preferred!r} is taken globally. "
            f"Re-run with --yes to accept {suggested!r}, or pass an explicit bucket name."
        )
    if ask_yes_no(f"Use suggested bucket name {suggested!r}?", default=True):
        return suggested
    while True:
        custom = ask(f"{label}: enter a globally unique bucket name", suggested)
        custom = normalize_s3_bucket_name(custom)
        custom_status, _ = resolve_s3_bucket_availability(custom, s3_client=s3_client)
        if custom_status == "available":
            return custom
        if custom_status == "owned":
            print(f"Bucket {custom!r} already exists in this account; will import it.")
            return custom
        print(f"Name {custom!r} is still taken globally. Try another name.")


def _prompt_globally_unique_cognito_prefix(
    preferred: str,
    account_id: str,
    region: str,
    *,
    interactive: bool,
    assume_yes: bool,
    cli_override: str = "",
    cognito_client: Any = None,
) -> str:
    from cdk_functions import resolve_cognito_domain_prefix_availability

    if cli_override:
        override = sanitize_cognito_domain_prefix(cli_override)
        validation_error = validate_cognito_domain_prefix(override)
        if validation_error:
            raise SystemExit(validation_error)
        if (
            resolve_cognito_domain_prefix_availability(
                override, region_name=region, cognito_client=cognito_client
            )
            != "available"
        ):
            raise SystemExit(
                f"Cognito domain prefix {override!r} is not available in {region} "
                "(likely taken by another AWS account). Pass a different "
                "--cognito-prefix."
            )
        return override

    preferred = sanitize_cognito_domain_prefix(preferred)
    availability = resolve_cognito_domain_prefix_availability(
        preferred, region_name=region, cognito_client=cognito_client
    )

    if availability == "taken":
        suggested = suggest_available_cognito_domain_prefix(
            preferred, account_id, region, cognito_client=cognito_client
        )
        print(
            f"Cognito hosted UI domain prefix {preferred!r} is not available in "
            f"{region} (likely taken by another AWS account)."
        )
        if not interactive:
            if assume_yes:
                print(f"Using suggested Cognito domain prefix: {suggested}")
                return suggested
            raise SystemExit(
                f"Cognito domain prefix {preferred!r} is not available in {region}. "
                f"Re-run with --yes to accept {suggested!r}, or pass --cognito-prefix."
            )
        if ask_yes_no(
            f"Use suggested Cognito domain prefix {suggested!r}?", default=True
        ):
            return suggested
        preferred = suggested

    if not interactive:
        print(f"Cognito domain prefix available: {preferred}")
        return preferred

    while True:
        raw = ask(
            "Cognito hosted UI domain prefix "
            "(must be globally unique in this region; "
            "cannot contain aws, amazon, or cognito)",
            preferred,
        )
        choice = sanitize_cognito_domain_prefix(raw)
        validation_error = validate_cognito_domain_prefix(choice)
        if validation_error:
            print(validation_error)
            continue
        if (
            resolve_cognito_domain_prefix_availability(
                choice, region_name=region, cognito_client=cognito_client
            )
            == "available"
        ):
            if choice != preferred:
                print(f"Using Cognito domain prefix: {choice}")
            else:
                print(f"Cognito domain prefix available: {choice}")
            return choice
        print(
            f"Prefix {choice!r} is not available in {region} "
            "(likely taken by another AWS account). Try another prefix."
        )


def resolve_globally_unique_install_names(
    answers: InstallAnswers,
    *,
    interactive: bool,
    assume_yes: bool,
    args: Optional[argparse.Namespace] = None,
) -> None:
    """Check S3 bucket and Cognito domain names; prompt for alternatives when taken."""
    import boto3

    args = args or argparse.Namespace()
    region = answers.aws_region
    account_id = answers.aws_account_id
    s3_client = boto3.client("s3", region_name=region)
    cognito_client = boto3.client("cognito-idp", region_name=region)

    defaults = derive_s3_bucket_names(answers.cdk_prefix)
    answers.s3_log_bucket_name = _prompt_globally_unique_s3_bucket(
        "S3 log/config bucket",
        defaults.get("S3_LOG_CONFIG_BUCKET_NAME", ""),
        account_id,
        interactive=interactive,
        assume_yes=assume_yes,
        cli_override=getattr(args, "s3_log_bucket", "") or "",
        s3_client=s3_client,
    )
    answers.s3_output_bucket_name = _prompt_globally_unique_s3_bucket(
        "S3 output bucket",
        defaults.get("S3_OUTPUT_BUCKET_NAME", ""),
        account_id,
        interactive=interactive,
        assume_yes=assume_yes,
        cli_override=getattr(args, "s3_output_bucket", "") or "",
        s3_client=s3_client,
    )

    if answers_use_headless(answers):
        answers.cognito_domain_prefix = ""
        return

    preferred_cognito = default_cognito_domain_prefix_from_cdk_prefix(
        answers.cdk_prefix
    )
    answers.cognito_domain_prefix = _prompt_globally_unique_cognito_prefix(
        preferred_cognito,
        account_id,
        region,
        interactive=interactive,
        assume_yes=assume_yes,
        cli_override=getattr(args, "cognito_prefix", "") or "",
        cognito_client=cognito_client,
    )


def validate_globally_unique_env_values(values: Dict[str, str]) -> List[str]:
    """Live AWS checks for globally unique resource names before deploy."""
    from cdk_functions import (
        resolve_cognito_domain_prefix_availability,
        resolve_s3_bucket_availability,
    )

    errors: List[str] = []
    region = (values.get("AWS_REGION") or "").strip()
    for env_key in ("S3_LOG_CONFIG_BUCKET_NAME", "S3_OUTPUT_BUCKET_NAME"):
        bucket_name = (values.get(env_key) or "").strip().lower()
        if not bucket_name:
            continue
        status, _ = resolve_s3_bucket_availability(bucket_name)
        if status == "globally_taken":
            errors.append(
                f"{env_key}={bucket_name!r} is taken globally by another AWS account. "
                "Re-run cdk_install.py to pick a unique name."
            )
    if values.get("ENABLE_HEADLESS_DEPLOYMENT") != "True":
        cognito_prefix = (values.get("COGNITO_USER_POOL_DOMAIN_PREFIX") or "").strip()
        if cognito_prefix and region:
            if (
                resolve_cognito_domain_prefix_availability(
                    cognito_prefix, region_name=region
                )
                == "taken"
            ):
                errors.append(
                    f"COGNITO_USER_POOL_DOMAIN_PREFIX={cognito_prefix!r} is not "
                    f"available in {region} (taken by another AWS account or existing "
                    "pool). Re-run cdk_install.py to pick a unique prefix."
                )
    return errors


def resolve_fixup_env_values(values: Dict[str, str]) -> Dict[str, str]:
    """Fill missing ECS cluster/service keys from CDK_PREFIX for post-deploy boto3 calls."""
    resolved = dict(values)
    for key, default in derive_ecs_resource_names(
        resolved.get("CDK_PREFIX", "")
    ).items():
        if not (resolved.get(key) or "").strip():
            resolved[key] = default
    for key, default in derive_s3_bucket_names(resolved.get("CDK_PREFIX", "")).items():
        if not (resolved.get(key) or "").strip():
            resolved[key] = default
    return resolved


def merge_preset(
    profile: str, overrides: Optional[Dict[str, str]] = None
) -> Dict[str, str]:
    base: Dict[str, str] = {}
    if profile == "demo":
        base.update(DEMO_PRESET)
    elif profile == "production":
        base.update(PRODUCTION_PRESET)
    elif profile == "headless":
        base.update(HEADLESS_PRESET)
    if overrides:
        base.update(overrides)
    return base


def answers_preset_profile(answers: "InstallAnswers") -> str:
    """Preset profile for env defaults (legacy ``headless`` maps to demo)."""
    if answers.profile == "headless":
        return "demo"
    return answers.profile


def answers_use_headless(answers: "InstallAnswers") -> bool:
    return answers.profile == "headless" or answers.enable_headless


def profile_allows_headless_add_on(answers: "InstallAnswers") -> bool:
    """Headless batch is incompatible with the Demonstration (Express) route."""
    if answers.profile == "demo":
        return False
    if answers.profile in ("production", "headless"):
        return True
    if answers.profile == "custom":
        return answers.custom_overrides.get("USE_ECS_EXPRESS_MODE") != "True"
    return False


def headless_profile_error(answers: "InstallAnswers") -> Optional[str]:
    if not answers_use_headless(answers):
        return None
    if answers.profile == "headless":
        return None
    if answers.profile == "demo":
        return (
            "Headless batch mode is not available with the Demonstration (Express) "
            "profile. Use the Headless shortcut profile, Production with headless, "
            "or Custom without ECS Express."
        )
    if (
        answers.profile == "custom"
        and answers.custom_overrides.get("USE_ECS_EXPRESS_MODE") == "True"
    ):
        return (
            "Headless batch mode requires USE_ECS_EXPRESS_MODE=False. "
            "Disable ECS Express in the custom profile or omit --headless."
        )
    return None


def validate_notify_email(email: str) -> Optional[str]:
    """Return an error message when email is not a plausible notification address."""
    cleaned = (email or "").strip()
    if not cleaned:
        return "Notification email address is required."
    if not re.match(r"^[^@\s]+@[^@\s]+\.[^@\s]+$", cleaned):
        return f"Invalid notification email address: {cleaned!r}"
    return None


DYNAMO_EXPORT_SCHEDULE_DAY_PRESETS = {
    "daily": "*",
    "weekdays": "MON-FRI",
    "weekends": "SAT-SUN",
}


def default_dynamo_export_s3_key() -> str:
    return "reports/dynamodb-usage/dynamodb_logs_export.csv"


def parse_schedule_time_hhmm(raw: str) -> Tuple[int, int]:
    match = re.match(r"^(\d{1,2}):(\d{2})$", (raw or "").strip())
    if not match:
        raise ValueError(
            "Schedule time must use HH:MM in 24-hour UTC format, e.g. 06:00."
        )
    hour, minute = int(match.group(1)), int(match.group(2))
    if hour > 23 or minute > 59:
        raise ValueError("Schedule time hour must be 0-23 and minute 0-59.")
    return hour, minute


def validate_schedule_time_hhmm(raw: str) -> Optional[str]:
    try:
        parse_schedule_time_hhmm(raw)
        return None
    except ValueError as exc:
        return str(exc)


def build_dynamo_export_cron_expression(
    hour: int, minute: int, days_preset: str
) -> str:
    """Build an EventBridge cron expression (UTC)."""
    dow = DYNAMO_EXPORT_SCHEDULE_DAY_PRESETS.get(
        (days_preset or "").strip().lower(), (days_preset or "*").strip().upper()
    )
    return f"cron({minute} {hour} ? * {dow} *)"


def prompt_dynamo_usage_log_export_options(
    answers: "InstallAnswers",
    args: argparse.Namespace,
    *,
    interactive: bool,
) -> None:
    if getattr(args, "dynamo_usage_log_export", False):
        answers.enable_dynamo_usage_log_export = True
    elif getattr(args, "no_dynamo_usage_log_export", False):
        answers.enable_dynamo_usage_log_export = False
    elif interactive:
        answers.enable_dynamo_usage_log_export = ask_yes_no(
            "Enable scheduled DynamoDB usage log export to S3 (EventBridge)?",
            default=False,
        )
    else:
        return

    if not answers.enable_dynamo_usage_log_export:
        return

    if getattr(args, "dynamo_export_schedule_time", "").strip():
        answers.dynamo_export_schedule_time = args.dynamo_export_schedule_time.strip()
    elif interactive:
        while True:
            raw = ask(
                "Export schedule time (UTC, HH:MM)",
                answers.dynamo_export_schedule_time,
            )
            err = validate_schedule_time_hhmm(raw)
            if not err:
                answers.dynamo_export_schedule_time = raw
                break
            print(err)

    if getattr(args, "dynamo_export_schedule_days", "").strip():
        answers.dynamo_export_schedule_days = (
            args.dynamo_export_schedule_days.strip().lower()
        )
    elif interactive:
        idx = ask_choice(
            "Run export on which days?",
            ["Every day", "Weekdays (Mon-Fri)", "Weekends (Sat-Sun)"],
            default_index=0,
        )
        answers.dynamo_export_schedule_days = ["daily", "weekdays", "weekends"][idx]

    default_key = answers.dynamo_export_s3_key or default_dynamo_export_s3_key()
    if getattr(args, "dynamo_export_s3_key", "").strip():
        answers.dynamo_export_s3_key = args.dynamo_export_s3_key.strip()
    elif interactive:
        answers.dynamo_export_s3_key = (
            ask("S3 object key for the CSV export", default_key).strip() or default_key
        )

    if getattr(args, "dynamo_export_date_attribute", "").strip():
        answers.dynamo_export_date_attribute = args.dynamo_export_date_attribute.strip()
    elif interactive:
        raw = ask(
            "DynamoDB date attribute for filtering",
            answers.dynamo_export_date_attribute,
        )
        if raw.strip():
            answers.dynamo_export_date_attribute = raw.strip()


def default_bedrock_invocation_log_group_name(cdk_prefix: str) -> str:
    """Default CloudWatch log group for Bedrock model invocation logging."""
    prefix = (cdk_prefix or "").strip()
    return f"/aws/bedrock/{prefix}model-invocations".rstrip("/")


def prompt_bedrock_model_invocation_logging_options(
    answers: "InstallAnswers",
    args: argparse.Namespace,
    *,
    interactive: bool,
) -> None:
    """
    Configure account/Region Bedrock model invocation logging to S3 + CloudWatch.

    Writes ENABLE_BEDROCK_MODEL_INVOCATION_LOGGING and related keys into cdk_config.env.
    """
    if getattr(args, "bedrock_model_invocation_logging", False):
        answers.enable_bedrock_model_invocation_logging = True
    elif getattr(args, "no_bedrock_model_invocation_logging", False):
        answers.enable_bedrock_model_invocation_logging = False
    elif interactive:
        default_on = answers_use_headless(answers)
        answers.enable_bedrock_model_invocation_logging = ask_yes_no(
            "Enable Amazon Bedrock model invocation logging "
            "(S3 log bucket + CloudWatch; account/Region-level setting)?",
            default=default_on,
        )
    else:
        return

    if not answers.enable_bedrock_model_invocation_logging:
        return

    default_prefix = answers.bedrock_invocation_s3_prefix or "bedrock-logs"
    if getattr(args, "bedrock_invocation_s3_prefix", "").strip():
        answers.bedrock_invocation_s3_prefix = (
            args.bedrock_invocation_s3_prefix.strip().strip("/")
        )
    elif interactive:
        answers.bedrock_invocation_s3_prefix = (
            ask(
                "S3 key prefix for Bedrock invocation logs (on the log/config bucket)",
                default_prefix,
            )
            .strip()
            .strip("/")
            or default_prefix
        )
    else:
        answers.bedrock_invocation_s3_prefix = default_prefix

    default_log_group = (
        answers.bedrock_invocation_log_group_name.strip()
        or default_bedrock_invocation_log_group_name(answers.cdk_prefix)
    )
    if getattr(args, "bedrock_invocation_log_group", "").strip():
        answers.bedrock_invocation_log_group_name = (
            args.bedrock_invocation_log_group.strip()
        )
    elif interactive:
        answers.bedrock_invocation_log_group_name = (
            ask(
                "CloudWatch log group for Bedrock invocation logs", default_log_group
            ).strip()
            or default_log_group
        )
    else:
        answers.bedrock_invocation_log_group_name = default_log_group

    default_retention = answers.bedrock_invocation_log_retention_days or "90"
    if getattr(args, "bedrock_invocation_log_retention_days", None) is not None:
        answers.bedrock_invocation_log_retention_days = str(
            args.bedrock_invocation_log_retention_days
        ).strip()
    elif interactive:
        while True:
            raw = ask(
                "CloudWatch log retention (days)",
                default_retention,
            ).strip()
            try:
                days = int(raw)
                if days <= 0:
                    raise ValueError
                answers.bedrock_invocation_log_retention_days = str(days)
                break
            except ValueError:
                print("Enter a positive whole number of days (e.g. 90).")
    else:
        answers.bedrock_invocation_log_retention_days = default_retention


def default_application_name(cdk_prefix: str) -> str:
    """Default Application tag value / Resource Group identity from CDK_PREFIX."""
    return f"{(cdk_prefix or '').strip()}llm-topic-modeller"


def prompt_application_tagging_and_appregistry_options(
    answers: "InstallAnswers",
    args: argparse.Namespace,
    *,
    interactive: bool,
) -> None:
    """
    Configure application tags / Resource Groups (preferred) and legacy AppRegistry.

    AppRegistry / myApplications closes to new customers after 30 Jul 2026.
    """
    if getattr(args, "appregistry", False):
        answers.enable_appregistry = True
    elif getattr(args, "no_appregistry", False):
        answers.enable_appregistry = False
    elif interactive:
        answers.enable_appregistry = ask_yes_no(
            "Enable legacy AppRegistry / myApplications stack "
            "(opt-in; new customers cut off after 30 Jul 2026)?",
            default=False,
        )
    # else: keep InstallAnswers default (False)

    if getattr(args, "application_resource_group", False):
        answers.enable_application_resource_group = True
    elif getattr(args, "no_application_resource_group", False):
        answers.enable_application_resource_group = False
    elif interactive:
        answers.enable_application_resource_group = ask_yes_no(
            "Tag all CDK-managed resources and create a Resource Group for this app "
            "(recommended for monitoring without AppRegistry)?",
            default=True,
        )
    # else: keep default True

    if not answers.enable_application_resource_group:
        return

    default_tag_key = (answers.application_tag_key or "Application").strip() or (
        "Application"
    )
    if getattr(args, "application_tag_key", "").strip():
        answers.application_tag_key = args.application_tag_key.strip()
    elif interactive:
        answers.application_tag_key = (
            ask("Application tag key", default_tag_key).strip() or default_tag_key
        )
    else:
        answers.application_tag_key = default_tag_key

    default_name = answers.application_name.strip() or default_application_name(
        answers.cdk_prefix
    )
    if getattr(args, "application_name", "").strip():
        answers.application_name = args.application_name.strip()
    elif interactive:
        answers.application_name = (
            ask(
                "Application name (tag value / Resource Group filter)", default_name
            ).strip()
            or default_name
        )
    else:
        answers.application_name = default_name


def validate_install_answers(answers: "InstallAnswers") -> List[str]:
    errors: List[str] = []
    msg = headless_profile_error(answers)
    if msg:
        errors.append(msg)
    if answers.enable_headless_output_notifications:
        if not answers_use_headless(answers):
            errors.append(
                "Headless output notifications require headless batch deployment."
            )
        email_error = validate_notify_email(answers.headless_output_notify_email)
        if email_error:
            errors.append(email_error)
    if answers.enable_dynamo_usage_log_export:
        time_error = validate_schedule_time_hhmm(answers.dynamo_export_schedule_time)
        if time_error:
            errors.append(time_error)
        days = (answers.dynamo_export_schedule_days or "").strip().lower()
        if days not in DYNAMO_EXPORT_SCHEDULE_DAY_PRESETS:
            errors.append(
                "DynamoDB export schedule days must be one of: "
                + ", ".join(DYNAMO_EXPORT_SCHEDULE_DAY_PRESETS)
            )
        if not (answers.dynamo_export_s3_key.strip() or default_dynamo_export_s3_key()):
            errors.append("DynamoDB export S3 object key is required.")
    if answers.enable_bedrock_model_invocation_logging:
        if not (answers.bedrock_invocation_s3_prefix or "").strip():
            errors.append(
                "Bedrock model invocation logging requires an S3 key prefix "
                "(bedrock_invocation_s3_prefix)."
            )
        log_group = (answers.bedrock_invocation_log_group_name or "").strip()
        if not log_group:
            errors.append(
                "Bedrock model invocation logging requires a CloudWatch log group name."
            )
        try:
            retention = int(
                (answers.bedrock_invocation_log_retention_days or "90").strip()
            )
            if retention <= 0:
                raise ValueError
        except ValueError:
            errors.append(
                "Bedrock invocation log retention days must be a positive integer."
            )
    if answers.enable_application_resource_group:
        app_name = answers.application_name.strip() or default_application_name(
            answers.cdk_prefix
        )
        if not app_name:
            errors.append(
                "Application Resource Group requires an application name "
                "(APPLICATION_NAME)."
            )
        if not (answers.application_tag_key or "").strip():
            errors.append("Application Resource Group requires an application tag key.")
    return errors


def answers_use_express_mode(answers: "InstallAnswers") -> bool:
    """True when the selected install profile/options enable ECS Express ingress."""
    if answers_use_headless(answers):
        return False
    preset = merge_preset(answers.profile, answers.custom_overrides)
    return preset.get("USE_ECS_EXPRESS_MODE") == "True"


def answers_use_public_subnets_only(answers: "InstallAnswers") -> bool:
    """Express ingress or demo-style headless batch (public subnets, no private install)."""
    if answers_use_express_mode(answers):
        return True
    if not answers_use_headless(answers):
        return False
    return answers_preset_profile(answers) in ("headless", "demo", "custom")


def build_env_values(answers: InstallAnswers) -> Dict[str, str]:
    cdk_folder = str(CDK_DIR).replace("\\", "/")
    if not cdk_folder.endswith("/"):
        cdk_folder += "/"

    s3_bucket_names = derive_s3_bucket_names(answers.cdk_prefix)
    if answers.s3_log_bucket_name:
        s3_bucket_names["S3_LOG_CONFIG_BUCKET_NAME"] = answers.s3_log_bucket_name
    if answers.s3_output_bucket_name:
        s3_bucket_names["S3_OUTPUT_BUCKET_NAME"] = answers.s3_output_bucket_name

    values: Dict[str, str] = merge_preset(
        answers_preset_profile(answers), answers.custom_overrides
    )
    values.update(
        {
            "CDK_PREFIX": answers.cdk_prefix,
            "AWS_REGION": answers.aws_region,
            "AWS_ACCOUNT_ID": answers.aws_account_id,
            "CDK_FOLDER": cdk_folder,
            **derive_ecs_resource_names(answers.cdk_prefix),
            **s3_bucket_names,
            "CONTEXT_FILE": "precheck.context.json",
            "COGNITO_USER_POOL_DOMAIN_PREFIX": sanitize_cognito_domain_prefix(
                answers.cognito_domain_prefix
            ),
            "GITHUB_REPO_BRANCH": answers.github_branch,
            "ECS_TASK_MEMORY_SIZE": answers.ecs_memory,
            "EXISTING_IGW_ID": answers.existing_igw_id,
            "EXISTING_LOAD_BALANCER_ARN": answers.existing_alb_arn,
            "EXISTING_LOAD_BALANCER_DNS": answers.existing_alb_dns
            or "placeholder_load_balancer_dns.net",
            "ENABLE_PI_AGENT_EXPRESS_SERVICE": (
                "True" if answers.enable_pi_express else "False"
            ),
            "ENABLE_PI_AGENT_ECS_SERVICE": (
                "True" if answers.enable_pi_legacy else "False"
            ),
            "ENABLE_ECS_SERVICE_CONNECT": (
                "True" if answers.enable_service_connect else "False"
            ),
            "ENABLE_S3_BATCH_ECS_TRIGGER": (
                "True"
                if answers.enable_s3_batch or answers_use_headless(answers)
                else "False"
            ),
            "ENABLE_HEADLESS_DEPLOYMENT": (
                "True" if answers_use_headless(answers) else "False"
            ),
        }
    )

    if answers_use_headless(answers):
        values.update(
            {
                "ENABLE_HEADLESS_DEPLOYMENT": "True",
                "ENABLE_S3_BATCH_ECS_TRIGGER": "True",
                "USE_ECS_EXPRESS_MODE": "False",
                "USE_CLOUDFRONT": "False",
                "RUN_USEAST_STACK": "False",
                "COGNITO_AUTH": "False",
                "ENABLE_PI_AGENT_EXPRESS_SERVICE": "False",
                "ENABLE_PI_AGENT_ECS_SERVICE": "False",
                "ENABLE_ECS_SERVICE_CONNECT": "False",
            }
        )
        if answers.enable_headless_output_notifications:
            iam_user = (
                answers.headless_output_iam_user_name.strip()
                or f"{answers.cdk_prefix}s3-output-reader"
            )
            values.update(
                {
                    "ENABLE_HEADLESS_OUTPUT_NOTIFICATIONS": "True",
                    "HEADLESS_OUTPUT_NOTIFY_EMAIL": (
                        answers.headless_output_notify_email.strip()
                    ),
                    "HEADLESS_OUTPUT_IAM_USER_NAME": iam_user,
                }
            )

    if answers.enable_dynamo_usage_log_export:
        hour, minute = parse_schedule_time_hhmm(answers.dynamo_export_schedule_time)
        s3_key = answers.dynamo_export_s3_key.strip() or default_dynamo_export_s3_key()
        values.update(
            {
                "ENABLE_DYNAMODB_USAGE_LOG_EXPORT": "True",
                "DYNAMODB_USAGE_LOG_EXPORT_SCHEDULE": build_dynamo_export_cron_expression(
                    hour, minute, answers.dynamo_export_schedule_days
                ),
                "DYNAMODB_USAGE_LOG_EXPORT_S3_KEY": s3_key,
                "DYNAMODB_USAGE_LOG_EXPORT_DATE_ATTRIBUTE": (
                    answers.dynamo_export_date_attribute.strip() or "timestamp"
                ),
            }
        )

    if answers.enable_bedrock_model_invocation_logging:
        values.update(
            {
                "ENABLE_BEDROCK_MODEL_INVOCATION_LOGGING": "True",
                "BEDROCK_MODEL_INVOCATION_S3_PREFIX": (
                    answers.bedrock_invocation_s3_prefix.strip().strip("/")
                    or "bedrock-logs"
                ),
                "BEDROCK_MODEL_INVOCATION_LOG_GROUP_NAME": (
                    answers.bedrock_invocation_log_group_name.strip()
                    or default_bedrock_invocation_log_group_name(answers.cdk_prefix)
                ),
                "BEDROCK_MODEL_INVOCATION_LOG_RETENTION_DAYS": (
                    answers.bedrock_invocation_log_retention_days.strip() or "90"
                ),
            }
        )

    # Application tags + Resource Groups (preferred); AppRegistry is legacy opt-in.
    # Always write tagging identity so Tags.of(app) has stable values even when
    # Resource Group creation is disabled.
    app_name = answers.application_name.strip() or default_application_name(
        answers.cdk_prefix
    )
    tag_key = (answers.application_tag_key or "Application").strip() or "Application"
    values["APPLICATION_NAME"] = app_name
    values["APPLICATION_TAG_KEY"] = tag_key
    values["APPLICATION_RESOURCE_GROUP_NAME"] = f"{app_name}-resources"
    values["ENABLE_APPLICATION_RESOURCE_GROUP"] = (
        "True" if answers.enable_application_resource_group else "False"
    )
    values["ENABLE_APPREGISTRY"] = "True" if answers.enable_appregistry else "False"
    if answers.enable_appregistry:
        values["APPREGISTRY_STACK_NAME"] = (
            f"{answers.cdk_prefix}{APPREGISTRY_STACK_SUFFIX}"
        )
        values["APPREGISTRY_APPLICATION_NAME"] = app_name

    use_cloudfront = values.get("USE_CLOUDFRONT") == "True"
    if answers.pi_enabled:
        values.update(
            {
                "PI_GRADIO_PORT": answers.pi_gradio_port,
                "ECS_SERVICE_CONNECT_DISCOVERY_NAME": answers.sc_discovery_name,
            }
        )
        if answers.enable_pi_express:
            values["ECS_EXPRESS_SC_PORT_NAME"] = "port-7860"
            values["ECS_PI_EXPRESS_SC_PORT_NAME"] = f"port-{answers.pi_gradio_port}"
        else:
            routing = answers.pi_alb_routing.strip().lower()
            path_prefix = normalize_pi_path_prefix(answers.pi_alb_path_prefix)
            priority = (
                answers.pi_alb_listener_rule_priority.strip()
                or default_pi_listener_priority(use_cloudfront)
            )
            values.update(
                {
                    "PI_ALB_ROUTING": routing,
                    "PI_ALB_PATH_PREFIX": path_prefix,
                    "PI_ALB_LISTENER_RULE_PRIORITY": priority,
                }
            )
            if routing in ("host", "both"):
                values["PI_ALB_HOST_HEADER"] = answers.pi_alb_host_header.strip()

    if answers.profile == "production":
        values["ACM_SSL_CERTIFICATE_ARN"] = answers.acm_cert_arn
        values["SSL_CERTIFICATE_DOMAIN"] = answers.ssl_domain
        values["CLOUDFRONT_DOMAIN"] = "cloudfront_placeholder.net"
        if answers.cloudfront_geo:
            values["CLOUDFRONT_GEO_RESTRICTION"] = answers.cloudfront_geo

    if answers.vpc_mode == "new":
        values["VPC_NAME"] = ""
        values["NEW_VPC_CIDR"] = answers.new_vpc_cidr
        values["NEW_VPC_DEFAULT_NAME"] = f"{answers.cdk_prefix}vpc"
        values["PUBLIC_SUBNETS_TO_USE"] = ""
        values["PRIVATE_SUBNETS_TO_USE"] = ""
        values["PUBLIC_SUBNET_CIDR_BLOCKS"] = ""
        values["PRIVATE_SUBNET_CIDR_BLOCKS"] = ""
        values["PUBLIC_SUBNET_AVAILABILITY_ZONES"] = ""
        values["PRIVATE_SUBNET_AVAILABILITY_ZONES"] = ""
    else:
        values["VPC_NAME"] = answers.vpc_name
        values["NEW_VPC_CIDR"] = ""
        apply_subnet_tier_env(
            values, answers, tier="public", mode=answers.public_subnet_mode
        )
        if answers_use_public_subnets_only(answers):
            values["ECS_EXPRESS_USE_PUBLIC_SUBNETS"] = "True"
            values["PRIVATE_SUBNETS_TO_USE"] = ""
            values["PRIVATE_SUBNET_CIDR_BLOCKS"] = ""
            values["PRIVATE_SUBNET_AVAILABILITY_ZONES"] = ""
        else:
            apply_subnet_tier_env(
                values, answers, tier="private", mode=answers.private_subnet_mode
            )

    # Pi agent is not supported in llm_topic_modeller yet (stack code retained for future).
    values["ENABLE_PI_AGENT_EXPRESS_SERVICE"] = "False"
    values["ENABLE_PI_AGENT_ECS_SERVICE"] = "False"

    return values


def validate_env_values(values: Dict[str, str]) -> List[str]:
    errors: List[str] = []

    express = values.get("USE_ECS_EXPRESS_MODE") == "True"
    acm = (values.get("ACM_SSL_CERTIFICATE_ARN") or "").strip()
    if express and acm:
        errors.append(
            "USE_ECS_EXPRESS_MODE=True cannot be used with ACM_SSL_CERTIFICATE_ARN."
        )

    if values.get("ENABLE_ECS_SERVICE_CONNECT") == "True" and express:
        errors.append(
            "ENABLE_ECS_SERVICE_CONNECT requires legacy Fargate (not Express)."
        )

    if values.get("ENABLE_S3_BATCH_ECS_TRIGGER") == "True" and express:
        errors.append(
            "ENABLE_S3_BATCH_ECS_TRIGGER requires legacy Fargate (not Express)."
        )

    headless = values.get("ENABLE_HEADLESS_DEPLOYMENT") == "True"
    if headless:
        if values.get("ENABLE_S3_BATCH_ECS_TRIGGER") != "True":
            errors.append(
                "ENABLE_HEADLESS_DEPLOYMENT requires ENABLE_S3_BATCH_ECS_TRIGGER=True."
            )
        if express:
            errors.append(
                "ENABLE_HEADLESS_DEPLOYMENT cannot be combined with USE_ECS_EXPRESS_MODE=True "
                "(use --profile headless, production --headless, or custom without Express)."
            )
        if values.get("USE_CLOUDFRONT") == "True":
            errors.append(
                "ENABLE_HEADLESS_DEPLOYMENT is incompatible with USE_CLOUDFRONT=True."
            )

    pi_ecs = values.get("ENABLE_PI_AGENT_ECS_SERVICE") == "True"
    pi_express = values.get("ENABLE_PI_AGENT_EXPRESS_SERVICE") == "True"
    if pi_ecs and pi_express:
        errors.append("Enable at most one agent mode (ECS or Express).")
    if pi_express and not express:
        errors.append(
            "ENABLE_PI_AGENT_EXPRESS_SERVICE requires USE_ECS_EXPRESS_MODE=True."
        )
    if pi_ecs and express:
        errors.append(
            "ENABLE_PI_AGENT_ECS_SERVICE requires USE_ECS_EXPRESS_MODE=False."
        )

    vpc_name = (values.get("VPC_NAME") or "").strip()
    new_cidr = (values.get("NEW_VPC_CIDR") or "").strip()
    if not vpc_name and not new_cidr:
        errors.append("Set VPC_NAME (existing VPC) or NEW_VPC_CIDR (new VPC).")
    elif new_cidr and not vpc_name:
        format_error = validate_new_vpc_cidr_format(new_cidr)
        if format_error:
            errors.append(format_error)

    if express:
        public_cidrs_raw = (values.get("PUBLIC_SUBNET_CIDR_BLOCKS") or "").strip()
        if public_cidrs_raw:
            try:
                import ast

                parsed = ast.literal_eval(public_cidrs_raw)
                public_cidrs = parsed if isinstance(parsed, list) else [str(parsed)]
            except (SyntaxError, ValueError):
                public_cidrs = [
                    part.strip().strip("'\"")
                    for part in public_cidrs_raw.split(",")
                    if part.strip()
                ]
            for cidr in public_cidrs:
                subnet_error = validate_public_subnet_cidr_for_express(str(cidr))
                if subnet_error:
                    errors.append(subnet_error)

    if values.get("USE_CLOUDFRONT") == "True" and not express:
        if not acm:
            errors.append(
                "Production CloudFront path requires ACM_SSL_CERTIFICATE_ARN."
            )
        if not (values.get("SSL_CERTIFICATE_DOMAIN") or "").strip():
            errors.append("Production CloudFront path requires SSL_CERTIFICATE_DOMAIN.")

    if not (values.get("CDK_PREFIX") or "").strip():
        errors.append("CDK_PREFIX is required.")
    prefix_lower = (values.get("CDK_PREFIX") or "").strip().lower()
    for env_key, bare_suffix in (
        ("S3_LOG_CONFIG_BUCKET_NAME", "s3-logs"),
        ("S3_OUTPUT_BUCKET_NAME", "s3-output"),
    ):
        bucket_name = (values.get(env_key) or "").strip().lower()
        if prefix_lower and bucket_name == bare_suffix:
            errors.append(
                f"{env_key} must include CDK_PREFIX (got bare '{bare_suffix}'; "
                f"expected '{prefix_lower}{bare_suffix}'). Re-run cdk_install or set "
                f"{env_key}={prefix_lower}{bare_suffix} in config/cdk_config.env."
            )
    if values.get("ENABLE_HEADLESS_DEPLOYMENT") != "True":
        cognito_prefix_error = validate_cognito_domain_prefix(
            values.get("COGNITO_USER_POOL_DOMAIN_PREFIX", "")
        )
        if cognito_prefix_error:
            errors.append(cognito_prefix_error)

    if pi_ecs and values.get("ENABLE_ECS_SERVICE_CONNECT") != "True":
        errors.append(
            "ENABLE_PI_AGENT_ECS_SERVICE=True requires ENABLE_ECS_SERVICE_CONNECT=True."
        )

    if headless and (
        pi_ecs or pi_express or values.get("ENABLE_ECS_SERVICE_CONNECT") == "True"
    ):
        errors.append(
            "ENABLE_HEADLESS_DEPLOYMENT is incompatible with agent mode or Service Connect."
        )

    if values.get("ENABLE_HEADLESS_OUTPUT_NOTIFICATIONS") == "True":
        if not headless:
            errors.append(
                "ENABLE_HEADLESS_OUTPUT_NOTIFICATIONS requires "
                "ENABLE_HEADLESS_DEPLOYMENT=True."
            )
        email_error = validate_notify_email(
            values.get("HEADLESS_OUTPUT_NOTIFY_EMAIL", "")
        )
        if email_error:
            errors.append(email_error)

    if values.get("ENABLE_DYNAMODB_USAGE_LOG_EXPORT") == "True":
        if values.get("SAVE_LOGS_TO_DYNAMODB") != "True":
            errors.append(
                "ENABLE_DYNAMODB_USAGE_LOG_EXPORT requires SAVE_LOGS_TO_DYNAMODB=True."
            )
        schedule = (values.get("DYNAMODB_USAGE_LOG_EXPORT_SCHEDULE") or "").strip()
        if not schedule.startswith("cron(") and not schedule.startswith("rate("):
            errors.append(
                "DYNAMODB_USAGE_LOG_EXPORT_SCHEDULE must be an EventBridge "
                f"cron(...) or rate(...) expression; got {schedule!r}."
            )
        if not (values.get("DYNAMODB_USAGE_LOG_EXPORT_S3_KEY") or "").strip():
            errors.append(
                "DYNAMODB_USAGE_LOG_EXPORT_S3_KEY is required when "
                "ENABLE_DYNAMODB_USAGE_LOG_EXPORT=True."
            )

    pi_routing = (values.get("PI_ALB_ROUTING") or "").strip().lower()
    if pi_ecs:
        if pi_routing not in PI_ALB_ROUTING_MODES:
            errors.append(
                f"PI_ALB_ROUTING must be one of {list(PI_ALB_ROUTING_MODES)}; got '{pi_routing}'."
            )
        elif (
            pi_routing in ("host", "both")
            and not (values.get("PI_ALB_HOST_HEADER") or "").strip()
        ):
            errors.append(
                "PI_ALB_HOST_HEADER is required when PI_ALB_ROUTING is 'host' or 'both'."
            )

    return errors


def build_app_config_env_values(values: Dict[str, str]) -> Dict[str, str]:
    """AWS deployment keys merged into config/app_config.env for ECS tasks."""
    prefix_lower = (values.get("CDK_PREFIX") or "").lower()
    headless = values.get("ENABLE_HEADLESS_DEPLOYMENT") == "True"
    pi_express = values.get("ENABLE_PI_AGENT_EXPRESS_SERVICE") == "True"

    def _name(env_key: str, suffix: str) -> str:
        return (values.get(env_key) or "").strip() or f"{prefix_lower}{suffix}"

    output_bucket = _name("S3_OUTPUT_BUCKET_NAME", "s3-output")

    return {
        "COGNITO_AUTH": "False" if headless or pi_express else "True",
        "RUN_AWS_FUNCTIONS": "True",
        "RUN_AWS_BEDROCK_MODELS": "True",
        "RUN_LOCAL_MODEL": "False",
        "RUN_GEMINI_MODELS": "False",
        "RUN_AZURE_MODELS": "False",
        "PRIORITISE_SSO_OVER_AWS_ENV_ACCESS_KEYS": "True",
        "DISPLAY_FILE_NAMES_IN_LOGS": "False",
        "SESSION_OUTPUT_FOLDER": "True",
        "SAVE_LOGS_TO_CSV": "True",
        "SAVE_LOGS_TO_DYNAMODB": "True",
        "SHOW_COSTS": "True",
        "S3_LOG_BUCKET": _name("S3_LOG_CONFIG_BUCKET_NAME", "s3-logs"),
        "S3_OUTPUTS_BUCKET": output_bucket if headless else "",
        "S3_OUTPUTS_FOLDER": "output/" if headless else "",
        "SAVE_OUTPUTS_TO_S3": "True" if headless else "False",
        "ACCESS_LOG_DYNAMODB_TABLE_NAME": _name(
            "ACCESS_LOG_DYNAMODB_TABLE_NAME", "dynamodb-access-logs"
        ),
        "FEEDBACK_LOG_DYNAMODB_TABLE_NAME": _name(
            "FEEDBACK_LOG_DYNAMODB_TABLE_NAME", "dynamodb-feedback-logs"
        ),
        "USAGE_LOG_DYNAMODB_TABLE_NAME": _name(
            "USAGE_LOG_DYNAMODB_TABLE_NAME", "dynamodb-usage-logs"
        ),
    }


def write_app_config_env_file(
    answers: InstallAnswers,
    values: Dict[str, str],
    *,
    overwrite: bool = False,
) -> Optional[Path]:
    if not answers.write_app_config_env:
        return None

    updates = build_app_config_env_values(values)
    APP_CONFIG_ENV_PATH.parent.mkdir(parents=True, exist_ok=True)

    if APP_CONFIG_ENV_PATH.is_file() and not overwrite:
        existing = read_env_file(APP_CONFIG_ENV_PATH)
        existing.update(updates)
        backup_file(APP_CONFIG_ENV_PATH)
        lines = [f"{k}={v}" for k, v in existing.items()]
        APP_CONFIG_ENV_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8")
        print(f"Updated {APP_CONFIG_ENV_PATH} (AWS deployment keys merged)")
        return APP_CONFIG_ENV_PATH

    if APP_CONFIG_ENV_EXAMPLE.is_file() and not overwrite:
        if APP_CONFIG_ENV_PATH.is_file():
            backup_file(APP_CONFIG_ENV_PATH)
        shutil.copy2(APP_CONFIG_ENV_EXAMPLE, APP_CONFIG_ENV_PATH)
        existing = read_env_file(APP_CONFIG_ENV_PATH)
        existing.update(updates)
        lines = [f"{k}={v}" for k, v in existing.items()]
        APP_CONFIG_ENV_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8")
        print(f"Created {APP_CONFIG_ENV_PATH} from example + AWS defaults")
        return APP_CONFIG_ENV_PATH

    if APP_CONFIG_ENV_PATH.is_file():
        backup_file(APP_CONFIG_ENV_PATH)
    lines = [
        "# Generated by cdk_install.py — app runtime config for AWS ECS",
        *[f"{k}={v}" for k, v in updates.items()],
    ]
    APP_CONFIG_ENV_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8")
    print(f"Wrote {APP_CONFIG_ENV_PATH}")
    return APP_CONFIG_ENV_PATH


def build_pi_agent_env_values(answers: InstallAnswers) -> Dict[str, str]:
    """Runtime settings for the Pi agent Gradio app (uploaded to S3 as pi_agent.env)."""
    values = {
        "PI_DEPLOYMENT_PROFILE": "aws-ecs",
        "PI_DEFAULT_PROVIDER": answers.pi_default_provider,
        "DOC_SUMMARISATION_GRADIO_URL": f"http://{answers.sc_discovery_name}:7860",
        "RUN_AWS_FUNCTIONS": "True",
        "AWS_REGION": answers.aws_region,
        "PI_GRADIO_PORT": answers.pi_gradio_port,
        "PI_DEFAULT_OCR_METHOD": "AWS Textract service - all PDF types",
        "PI_DEFAULT_PII_METHOD": "AWS Comprehend",
    }
    if answers.enable_pi_express:
        values["RUN_FASTAPI"] = "True"
        return values
    path_prefix = normalize_pi_path_prefix(answers.pi_alb_path_prefix)
    if answers.pi_alb_routing.strip().lower() in ("path", "both"):
        values["PI_ROOT_PATH"] = path_prefix
        values["ROOT_PATH"] = path_prefix
        values["FASTAPI_ROOT_PATH"] = path_prefix
    return values


def write_pi_agent_env_file(
    answers: InstallAnswers,
    *,
    overwrite: bool = False,
) -> Optional[Path]:
    if not answers.pi_enabled or not answers.write_pi_agent_env:
        return None

    updates = build_pi_agent_env_values(answers)
    PI_AGENT_ENV_PATH.parent.mkdir(parents=True, exist_ok=True)

    if PI_AGENT_ENV_PATH.is_file() and not overwrite:
        existing = read_env_file(PI_AGENT_ENV_PATH)
        existing.update(updates)
        backup_file(PI_AGENT_ENV_PATH)
        lines = [f"{k}={v}" for k, v in existing.items()]
        PI_AGENT_ENV_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8")
        print(f"Updated {PI_AGENT_ENV_PATH} (AWS/Pi agent keys merged)")
        return PI_AGENT_ENV_PATH

    if PI_AGENT_ENV_EXAMPLE.is_file() and not overwrite:
        backup_file(PI_AGENT_ENV_PATH) if PI_AGENT_ENV_PATH.is_file() else None
        shutil.copy2(PI_AGENT_ENV_EXAMPLE, PI_AGENT_ENV_PATH)
        existing = read_env_file(PI_AGENT_ENV_PATH)
        existing.update(updates)
        lines = [f"{k}={v}" for k, v in existing.items()]
        PI_AGENT_ENV_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8")
        print(f"Created {PI_AGENT_ENV_PATH} from example + AWS defaults")
        return PI_AGENT_ENV_PATH

    backup_file(PI_AGENT_ENV_PATH) if PI_AGENT_ENV_PATH.is_file() else None
    lines = [
        "# Generated by cdk_install.py — Pi agent runtime config for AWS ECS",
        *[f"{k}={v}" for k, v in updates.items()],
    ]
    PI_AGENT_ENV_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8")
    print(f"Wrote {PI_AGENT_ENV_PATH}")
    return PI_AGENT_ENV_PATH


def write_env_file(path: Path, values: Dict[str, str]) -> Path:
    backup_file(path)
    lines = [f"{key}={val}" for key, val in values.items()]
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
    print(f"Wrote {path}")
    return path


def read_env_file(path: Path) -> Dict[str, str]:
    values: Dict[str, str] = {}
    if not path.is_file():
        return values
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        if "=" in line:
            key, _, val = line.partition("=")
            values[key.strip()] = val.strip()
    return values


def patch_env_file(path: Path, updates: Dict[str, str]) -> None:
    values = read_env_file(path)
    values.update(updates)
    write_env_file(path, values)


def print_summary(values: Dict[str, str], python_exe: Optional[Path] = None) -> None:
    keys = [
        "CDK_PREFIX",
        "AWS_REGION",
        "AWS_ACCOUNT_ID",
        "USE_ECS_EXPRESS_MODE",
        "USE_CLOUDFRONT",
        "ENABLE_HEADLESS_DEPLOYMENT",
        "ENABLE_S3_BATCH_ECS_TRIGGER",
        "ENABLE_BEDROCK_MODEL_INVOCATION_LOGGING",
        "ENABLE_APPREGISTRY",
        "ENABLE_APPLICATION_RESOURCE_GROUP",
        "APPLICATION_NAME",
        "ENABLE_RESOURCE_DELETE_PROTECTION",
        "VPC_NAME",
        "NEW_VPC_CIDR",
        "ACM_SSL_CERTIFICATE_ARN",
        "SSL_CERTIFICATE_DOMAIN",
        "ENABLE_PI_AGENT_EXPRESS_SERVICE",
        "ENABLE_PI_AGENT_ECS_SERVICE",
    ]
    print("\n--- Configuration summary ---")
    for key in keys:
        if key in values:
            print(f"  {key}={values[key]}")
    if python_exe:
        print(f"  cdk.json app={format_cdk_app_command(python_exe)}")
    print("----------------------------\n")


# ---------------------------------------------------------------------------
# Deploy runner
# ---------------------------------------------------------------------------


def _deploy_env() -> Dict[str, str]:
    return build_cdk_subprocess_env()


def run_cdk_command(
    args: List[str], *, check: bool = True
) -> subprocess.CompletedProcess:
    cdk_exe = resolve_cdk_executable()
    cmd = [cdk_exe, *args]
    print(f"Running: {' '.join(cmd)}")
    return subprocess.run(
        cmd,
        cwd=str(CDK_DIR),
        env=_deploy_env(),
        check=check,
    )


@dataclass(frozen=True)
class ExistingStack:
    name: str
    region: str
    status: str
    termination_protection: bool = False


def _should_check_cloudfront_stack(
    config_values: Optional[Dict[str, str]] = None,
) -> bool:
    """Skip us-east-1 CloudFront stack lookup when config disables that path."""
    if not config_values:
        return True
    use_cloudfront = config_values.get("USE_CLOUDFRONT")
    run_useast = config_values.get("RUN_USEAST_STACK")
    if use_cloudfront is not None and use_cloudfront != "True":
        return False
    if run_useast is not None and run_useast != "True":
        return False
    return True


def _stack_check_skippable_error(exc: Exception) -> bool:
    """Permission / SCP errors in one region must not block checks in others."""
    from botocore.exceptions import ClientError

    if not isinstance(exc, ClientError):
        return False
    code = exc.response.get("Error", {}).get("Code", "")
    return code in (
        "AccessDenied",
        "UnauthorizedOperation",
        "AuthorizationError",
        "AccessDeniedException",
    )


def derived_appregistry_stack_name(
    config_values: Optional[Dict[str, str]] = None,
) -> Optional[str]:
    """Resolve AppRegistry stack name from config (explicit or CDK_PREFIX default)."""
    if not config_values:
        return None
    explicit = (config_values.get("APPREGISTRY_STACK_NAME") or "").strip()
    if explicit:
        return explicit
    prefix = (config_values.get("CDK_PREFIX") or "").strip()
    if prefix:
        return f"{prefix}{APPREGISTRY_STACK_SUFFIX}"
    return None


def list_regional_appregistry_stack_names(region: str) -> List[str]:
    """List active AppRegistry stacks in the regional account (suffix match)."""
    import boto3
    from botocore.exceptions import ClientError

    cfn = boto3.client("cloudformation", region_name=region)
    skip_statuses = {"DELETE_COMPLETE", "DELETE_IN_PROGRESS"}
    names: List[str] = []
    try:
        paginator = cfn.get_paginator("list_stacks")
        for page in paginator.paginate():
            for summary in page.get("StackSummaries", []):
                name = summary.get("StackName", "")
                status = summary.get("StackStatus", "")
                if not name.endswith(APPREGISTRY_STACK_SUFFIX):
                    continue
                if status in skip_statuses:
                    continue
                names.append(name)
    except ClientError as exc:
        if _stack_check_skippable_error(exc):
            code = exc.response.get("Error", {}).get("Code", "ClientError")
            print(
                f"Warning: could not list AppRegistry stacks in {region} "
                f"({code}). Continuing with configured stack names only."
            )
            return []
        raise
    return sorted(dict.fromkeys(names))


def appregistry_stack_names_to_check(
    config_values: Optional[Dict[str, str]] = None,
    *,
    discovered_names: Optional[Sequence[str]] = None,
) -> List[str]:
    """Merge configured and discovered AppRegistry stack names (deduped)."""
    names: List[str] = []
    derived = derived_appregistry_stack_name(config_values)
    if derived:
        names.append(derived)
    for name in discovered_names or ():
        if name not in names:
            names.append(name)
    return names


def stacks_to_check(
    regional_region: str,
    config_values: Optional[Dict[str, str]] = None,
    *,
    discovered_appregistry_names: Optional[Sequence[str]] = None,
) -> List[Tuple[str, str]]:
    """Return (stack_name, region) pairs in safe deletion order."""
    checks: List[Tuple[str, str]] = []
    if _should_check_cloudfront_stack(config_values):
        checks.append((CLOUDFRONT_STACK, CLOUDFRONT_STACK_REGION))
    for appregistry_name in appregistry_stack_names_to_check(
        config_values,
        discovered_names=discovered_appregistry_names,
    ):
        checks.append((appregistry_name, regional_region))
    checks.append((REGIONAL_STACK, regional_region))
    return checks


def describe_existing_stack(stack_name: str, region: str) -> Optional[ExistingStack]:
    import boto3
    from botocore.exceptions import ClientError

    cfn = boto3.client("cloudformation", region_name=region)
    try:
        response = cfn.describe_stacks(StackName=stack_name)
    except ClientError as exc:
        code = exc.response.get("Error", {}).get("Code", "")
        if code in ("ValidationError", "ResourceNotFoundException") or (
            "does not exist" in str(exc)
        ):
            return None
        raise
    stack = response["Stacks"][0]
    status = stack.get("StackStatus", "")
    if status == "DELETE_COMPLETE":
        return None
    return ExistingStack(
        name=stack_name,
        region=region,
        status=status,
        termination_protection=bool(stack.get("EnableTerminationProtection", False)),
    )


def discover_existing_llm_topic_stacks(
    regional_region: str,
    config_values: Optional[Dict[str, str]] = None,
) -> List[ExistingStack]:
    """Find deployed llm_topic_modeller CloudFormation stacks in the target account."""
    from botocore.exceptions import ClientError

    discovered_appregistry = list_regional_appregistry_stack_names(regional_region)
    checks = stacks_to_check(
        regional_region,
        config_values,
        discovered_appregistry_names=discovered_appregistry,
    )
    ordered: List[ExistingStack] = []
    for stack_name, region in checks:
        try:
            info = describe_existing_stack(stack_name, region)
        except ClientError as exc:
            if _stack_check_skippable_error(exc):
                code = exc.response.get("Error", {}).get("Code", "ClientError")
                print(
                    f"Warning: could not check stack {stack_name!r} in {region} "
                    f"({code}). Continuing with other regions."
                )
                continue
            raise
        if info:
            ordered.append(info)
    return ordered


def discover_existing_doc_summarisation_stacks(
    regional_region: str,
    config_values: Optional[Dict[str, str]] = None,
) -> List[ExistingStack]:
    """Deprecated alias for discover_existing_llm_topic_stacks."""
    return discover_existing_llm_topic_stacks(regional_region, config_values)


def force_delete_cloudformation_stacks(
    stacks: Sequence[ExistingStack],
    *,
    wait: bool = True,
) -> None:
    """Delete stacks via CloudFormation (disables termination protection first)."""
    import boto3
    from botocore.exceptions import ClientError

    clients: Dict[str, Any] = {}

    def cfn_for(region: str):
        if region not in clients:
            clients[region] = boto3.client("cloudformation", region_name=region)
        return clients[region]

    for stack in stacks:
        cfn = cfn_for(stack.region)
        if stack.termination_protection:
            print(f"Disabling termination protection on {stack.name} ({stack.region})")
            cfn.update_termination_protection(
                EnableTerminationProtection=False,
                StackName=stack.name,
            )
        if stack.status == "DELETE_IN_PROGRESS":
            print(f"{stack.name} ({stack.region}) is already deleting.")
        else:
            print(f"Deleting stack {stack.name} ({stack.region}) ...")
            try:
                cfn.delete_stack(StackName=stack.name)
            except ClientError as exc:
                code = exc.response.get("Error", {}).get("Code", "")
                if code == "ValidationError" and "DELETE_IN_PROGRESS" in str(exc):
                    print(f"{stack.name} is already deleting.")
                else:
                    raise

    if not wait:
        return

    for stack in stacks:
        cfn = cfn_for(stack.region)
        print(f"Waiting for {stack.name} ({stack.region}) to finish deleting ...")
        try:
            cfn.get_waiter("stack_delete_complete").wait(
                StackName=stack.name,
                WaiterConfig={"Delay": 15, "MaxAttempts": 120},
            )
        except ClientError:
            detail = describe_existing_stack(stack.name, stack.region)
            status = detail.status if detail else "unknown"
            raise SystemExit(
                f"Stack {stack.name} ({stack.region}) did not delete cleanly "
                f"(status={status}). Resolve in the CloudFormation console and retry."
            ) from None


def handle_existing_stacks_at_start(
    args: argparse.Namespace,
    regional_region: str,
    *,
    config_values: Optional[Dict[str, str]] = None,
) -> None:
    """At wizard start: report existing stacks and optionally force-delete them."""
    if getattr(args, "skip_stack_check", False) or args.config_only or args.synth_only:
        return

    if config_values is None and ENV_PATH.is_file():
        config_values = read_env_file(ENV_PATH)

    try:
        existing = discover_existing_llm_topic_stacks(regional_region, config_values)
    except Exception as exc:
        print(f"Existing stack check skipped: {exc}")
        return

    if not existing:
        return

    print("\n--- Existing llm_topic_modeller CloudFormation stacks ---")
    for stack in existing:
        line = f"  {stack.name} ({stack.region}): {stack.status}"
        if stack.termination_protection:
            line += " [termination protection ON]"
        print(line)

    if args.force_delete_stacks:
        should_delete = True
    elif args.yes:
        print(
            "\nStacks already exist in this account/region. "
            "Pass --force-delete-stacks to remove them before deploy, "
            "or omit it to update in place."
        )
        return
    else:
        should_delete = ask_yes_no(
            "Force-delete these stacks before continuing? "
            "(disables termination protection and deletes all stack resources)",
            default=False,
        )

    if not should_delete:
        print("Keeping existing stacks (deploy will update them in place).\n")
        return

    if not args.force_delete_stacks and not args.yes:
        if not ask_yes_no(
            "This permanently deletes AWS resources in these stacks. Proceed?",
            default=False,
        ):
            print("Stack deletion cancelled.\n")
            return

    force_delete_cloudformation_stacks(existing)
    print("Existing stacks deleted.\n")


def fetch_stack_output(
    stack_name: str,
    output_key: str,
    region: str,
) -> Optional[str]:
    import boto3
    from botocore.exceptions import ClientError

    cfn = boto3.client("cloudformation", region_name=region)
    try:
        response = cfn.describe_stacks(StackName=stack_name)
    except ClientError:
        return None
    for stack in response.get("Stacks", []):
        for output in stack.get("Outputs", []):
            if output.get("OutputKey") == output_key:
                return output.get("OutputValue")
    return None


def apply_post_deploy_fixup(values: Dict[str, str], assume_yes: bool) -> bool:
    """Return True if post-deploy fixup changed env and/or Cognito callbacks."""
    from cdk_post_deploy import (
        apply_cognito_alb_callback_fixup,
        cognito_alb_callbacks_need_update,
    )

    if values.get("ENABLE_HEADLESS_DEPLOYMENT") == "True":
        return False

    region = values.get("AWS_REGION", "")
    express = values.get("USE_ECS_EXPRESS_MODE") == "True"
    cloudfront = values.get("USE_CLOUDFRONT") == "True"
    fixup_applied = False

    pool_id = fetch_stack_output(REGIONAL_STACK, "CognitoPoolId", region)
    client_id = fetch_stack_output(REGIONAL_STACK, "CognitoAppClientId", region)
    if not pool_id or not client_id:
        print(
            "CognitoPoolId or CognitoAppClientId stack output missing; "
            "skipping Cognito callback fixup."
        )
        return False

    if express and not cloudfront:
        from cdk_config import normalize_https_redirect_url

        endpoint = fetch_stack_output(REGIONAL_STACK, "ExpressServiceEndpoint", region)
        if not endpoint:
            print("No ExpressServiceEndpoint output found; skipping Cognito URL fixup.")
            return False
        endpoint = normalize_https_redirect_url(endpoint)
        current = (values.get("ECS_EXPRESS_COGNITO_REDIRECT_BASE") or "").strip()
        env_needs_patch = current != endpoint
        cognito_needs_update = cognito_alb_callbacks_need_update(
            pool_id, client_id, endpoint, aws_region=region
        )
        if not env_needs_patch and not cognito_needs_update:
            print("Express Cognito redirect base and callback URLs already set.")
        else:
            if env_needs_patch:
                print(f"Setting ECS_EXPRESS_COGNITO_REDIRECT_BASE={endpoint}")
                patch_env_file(
                    ENV_PATH,
                    {
                        "ECS_EXPRESS_COGNITO_REDIRECT_BASE": endpoint,
                        "COGNITO_REDIRECTION_URL": endpoint,
                    },
                )
                fixup_applied = True
            if cognito_needs_update and (
                assume_yes
                or ask_yes_no(
                    "Update Cognito app client callback URLs for the Express endpoint "
                    "(no CDK redeploy)?",
                    True,
                )
            ):
                if apply_cognito_alb_callback_fixup(
                    user_pool_id=pool_id,
                    client_id=client_id,
                    redirect_base=endpoint,
                    aws_region=region,
                ):
                    fixup_applied = True
            elif env_needs_patch and cognito_needs_update:
                print(
                    "Env updated with Express URL; Cognito callbacks unchanged. "
                    "ALB login will fail until callback URLs are updated."
                )

    elif cloudfront:
        cf_domain = fetch_stack_output(
            CLOUDFRONT_STACK, "CloudFrontDistributionURL", CLOUDFRONT_STACK_REGION
        )
        if not cf_domain:
            print(
                "No CloudFrontDistributionURL output found; skipping CloudFront fixup."
            )
            return fixup_applied
        redirect_base = f"https://{cf_domain.strip()}"
        current = (values.get("CLOUDFRONT_DOMAIN") or "").strip()
        env_needs_patch = current != cf_domain.strip()
        cognito_needs_update = cognito_alb_callbacks_need_update(
            pool_id, client_id, redirect_base, aws_region=region
        )
        if not env_needs_patch and not cognito_needs_update:
            print("CloudFront domain and Cognito callback URLs already set.")
            return fixup_applied
        if env_needs_patch:
            print(f"Setting CLOUDFRONT_DOMAIN={cf_domain}")
            patch_env_file(
                ENV_PATH,
                {
                    "CLOUDFRONT_DOMAIN": cf_domain,
                    "COGNITO_REDIRECTION_URL": redirect_base,
                },
            )
            fixup_applied = True
        if cognito_needs_update and (
            assume_yes
            or ask_yes_no(
                "Update Cognito app client callback URLs for the CloudFront domain "
                "(no CDK redeploy)?",
                True,
            )
        ):
            if apply_cognito_alb_callback_fixup(
                user_pool_id=pool_id,
                client_id=client_id,
                redirect_base=redirect_base,
                aws_region=region,
            ):
                fixup_applied = True
        elif env_needs_patch and cognito_needs_update:
            print(
                "Env updated with CloudFront domain; Cognito callbacks unchanged. "
                "ALB login will fail until callback URLs are updated."
            )

    if express:
        from cdk_post_deploy import apply_cognito_secret_fixup_from_stack

        resolved = resolve_fixup_env_values(values)
        cluster_name = resolved["CLUSTER_NAME"]
        main_service = resolved["ECS_EXPRESS_SERVICE_NAME"]
        secret_name = resolved.get("COGNITO_USER_POOL_CLIENT_SECRET_NAME", "")
        if not main_service:
            print(
                "ECS express service name could not be resolved; "
                "skipping Express Cognito secret sync."
            )
            return fixup_applied
        try:
            if pool_id and client_id:
                if apply_cognito_secret_fixup_from_stack(
                    stack_name=REGIONAL_STACK,
                    secret_name=secret_name,
                    cluster_name=cluster_name,
                    main_service_name=main_service,
                    aws_region=region,
                    recycle_tasks=False,
                ):
                    fixup_applied = True
        except Exception as exc:
            print(f"Warning: could not sync Express Cognito secret: {exc}")

    return fixup_applied


def run_quickstart(python_exe: Path) -> None:
    if not QUICKSTART_SCRIPT.is_file():
        raise SystemExit(f"Quickstart script not found: {QUICKSTART_SCRIPT}")
    cmd = [str(python_exe), str(QUICKSTART_SCRIPT)]
    print(f"Running: {' '.join(cmd)}")
    subprocess.run(cmd, cwd=str(CDK_DIR), env=_deploy_env(), check=True)


# ---------------------------------------------------------------------------
# Pi agent configuration (disabled in installer; stack code retained for future)
# ---------------------------------------------------------------------------

_DEPRECATED_PI_CLI_FLAGS = (
    "--enable-pi",
    "--enable-pi-express",
    "--enable-pi-legacy",
    "--pi-alb-routing",
    "--pi-path-prefix",
    "--pi-host-header",
    "--pi-listener-priority",
    "--pi-gradio-port",
    "--sc-discovery-name",
    "--pi-provider",
    "--skip-pi-agent-env",
)


def warn_deprecated_pi_cli_flags(args: argparse.Namespace) -> None:
    """Ignore Pi agent CLI flags — not supported in llm_topic_modeller yet."""
    argv = getattr(args, "_argv", ()) or ()
    if any(
        token == flag or token.startswith(f"{flag}=")
        for flag in _DEPRECATED_PI_CLI_FLAGS
        for token in argv
    ):
        print(
            "Note: Pi agent is not supported in llm_topic_modeller yet; "
            "ignoring Pi-related flags."
        )
    answers_pi = (
        getattr(args, "enable_pi", False)
        or getattr(args, "enable_pi_express", False)
        or getattr(args, "enable_pi_legacy", False)
    )
    if answers_pi:
        print(
            "Note: Pi agent is not supported in llm_topic_modeller yet; "
            "ignoring Pi-related flags."
        )


def apply_pi_cli_flags(args: argparse.Namespace, answers: InstallAnswers) -> None:
    if getattr(args, "enable_pi", False):
        preset = merge_preset(answers.profile, answers.custom_overrides)
        if preset.get("USE_ECS_EXPRESS_MODE") == "True":
            answers.enable_pi_express = True
        else:
            answers.enable_pi_legacy = True
            answers.enable_service_connect = True
    if getattr(args, "enable_pi_express", False):
        answers.enable_pi_express = True
    if getattr(args, "enable_pi_legacy", False):
        answers.enable_pi_legacy = True
        answers.enable_service_connect = True
    if args.pi_alb_routing:
        answers.pi_alb_routing = args.pi_alb_routing
    if args.pi_path_prefix:
        answers.pi_alb_path_prefix = args.pi_path_prefix
    if args.pi_host_header:
        answers.pi_alb_host_header = args.pi_host_header
    if args.pi_listener_priority:
        answers.pi_alb_listener_rule_priority = args.pi_listener_priority
    if args.pi_gradio_port:
        answers.pi_gradio_port = args.pi_gradio_port
    if args.sc_discovery_name:
        answers.sc_discovery_name = args.sc_discovery_name
    if args.pi_provider:
        answers.pi_default_provider = args.pi_provider
    if args.skip_pi_agent_env:
        answers.write_pi_agent_env = False


def configure_app_config_options(
    answers: InstallAnswers,
    args: argparse.Namespace,
    *,
    interactive: bool,
) -> None:
    if getattr(args, "skip_app_config_env", False):
        answers.write_app_config_env = False
        return

    if getattr(args, "overwrite_app_config_env", False):
        answers.write_app_config_env = True
        answers.overwrite_app_config_env = True
        return

    if not interactive:
        answers.write_app_config_env = True
        return

    example_label = (
        APP_CONFIG_ENV_EXAMPLE.name if APP_CONFIG_ENV_EXAMPLE.is_file() else "defaults"
    )
    if ask_yes_no(
        f"Write/update {APP_CONFIG_ENV_PATH.name} for the deployed app "
        f"(from {example_label} + AWS resource names)?",
        default=True,
    ):
        answers.write_app_config_env = True
        if APP_CONFIG_ENV_PATH.is_file():
            answers.overwrite_app_config_env = ask_yes_no(
                f"{APP_CONFIG_ENV_PATH.name} exists — replace file from example template?",
                default=False,
            )
    else:
        answers.write_app_config_env = False


def configure_pi_options(
    answers: InstallAnswers,
    args: argparse.Namespace,
    *,
    interactive: bool,
    assume_yes: bool,
) -> None:
    preset = merge_preset(answers.profile, answers.custom_overrides)
    use_express = preset.get("USE_ECS_EXPRESS_MODE") == "True"
    use_cloudfront = preset.get("USE_CLOUDFRONT") == "True"

    apply_pi_cli_flags(args, answers)

    if not answers.pi_enabled and interactive:
        if use_express:
            label = (
                "Deploy agent mode (second Gradio app on Express, dedicated HTTPS URL)?"
            )
            answers.enable_pi_express = ask_yes_no(label, default=False)
        else:
            label = "Deploy agent mode (second Gradio app on legacy Fargate + Service Connect)?"
            if ask_yes_no(label, default=False):
                answers.enable_pi_legacy = True
                answers.enable_service_connect = True

    if not answers.pi_enabled:
        return

    if not answers.pi_alb_listener_rule_priority and not answers.enable_pi_express:
        answers.pi_alb_listener_rule_priority = default_pi_listener_priority(
            use_cloudfront
        )

    if answers.enable_pi_express:
        if interactive and not assume_yes:
            print("\n--- Agent mode (ECS Express) ---")
            answers.sc_discovery_name = ask(
                "Service Connect discovery name for main app",
                answers.sc_discovery_name,
            )
            answers.pi_gradio_port = ask("Agent Gradio port", answers.pi_gradio_port)
            if ask_yes_no(
                f"Write/update {PI_AGENT_ENV_PATH.name} for AWS ECS (DOC_SUMMARISATION_GRADIO_URL, etc.)?",
                default=True,
            ):
                answers.write_pi_agent_env = True
                if PI_AGENT_ENV_PATH.is_file():
                    answers.overwrite_pi_agent_env = ask_yes_no(
                        "pi_agent.env exists — replace file from example template?",
                        default=False,
                    )
            else:
                answers.write_pi_agent_env = False
        print(
            f"Agent mode: Express (dedicated HTTPS endpoint per service); "
            f"Service Connect discovery={answers.sc_discovery_name}"
        )
        return

    if interactive and not (
        args.pi_alb_routing or args.pi_path_prefix or args.pi_host_header or assume_yes
    ):
        print("\n--- Agent mode ALB routing ---")
        ridx = ask_choice(
            "How should the shared ALB route traffic to the Agent UI?",
            [
                "Path prefix (default /agent/ — e.g. https://host/agent/)",
                "Dedicated hostname (PI_ALB_HOST_HEADER)",
                "Both path prefix and hostname",
            ],
            default_index=0,
        )
        answers.pi_alb_routing = PI_ALB_ROUTING_MODES[ridx]

        if answers.pi_alb_routing in ("path", "both"):
            answers.pi_alb_path_prefix = ask(
                "Agent path prefix",
                answers.pi_alb_path_prefix,
            )
        if answers.pi_alb_routing in ("host", "both"):
            default_host = ""
            if answers.ssl_domain:
                default_host = f"agent.{answers.ssl_domain}"
            answers.pi_alb_host_header = ask(
                "Agent ALB host header (DNS CNAME to CloudFront/ALB)",
                default_host,
            )

        if use_cloudfront:
            default_pri = default_pi_listener_priority(True)
            answers.pi_alb_listener_rule_priority = ask(
                "Agent ALB listener rule priority (default 3; priorities 1–2 reserved)",
                default_pri,
            )

        answers.sc_discovery_name = ask(
            "Service Connect discovery name for main app",
            answers.sc_discovery_name,
        )
        answers.pi_gradio_port = ask("Agent Gradio port", answers.pi_gradio_port)

        if ask_yes_no(
            f"Write/update {PI_AGENT_ENV_PATH.name} for AWS ECS (DOC_SUMMARISATION_GRADIO_URL, etc.)?",
            default=True,
        ):
            answers.write_pi_agent_env = True
            if PI_AGENT_ENV_PATH.is_file():
                answers.overwrite_pi_agent_env = ask_yes_no(
                    "pi_agent.env exists — replace file from example template?",
                    default=False,
                )
        else:
            answers.write_pi_agent_env = False

    print(
        f"Agent mode: "
        f"{'Express' if answers.enable_pi_express else 'legacy Fargate'}, "
        f"routing={answers.pi_alb_routing}, "
        f"prefix={normalize_pi_path_prefix(answers.pi_alb_path_prefix)}"
    )


# ---------------------------------------------------------------------------
# Interactive wizard
# ---------------------------------------------------------------------------


def run_wizard(args: argparse.Namespace) -> InstallAnswers:
    answers = InstallAnswers()
    interactive = not args.yes
    assume_yes = args.yes

    if args.profile:
        answers.profile = args.profile
    elif interactive:
        idx = ask_choice(
            "Deployment profile",
            [
                "Demonstration (Express, no CloudFront, no delete protection)",
                "Production (ACM cert, CloudFront, delete protection)",
                "Headless batch only (demo-style networking, no Express web UI)",
                "Custom (configure individual toggles)",
            ],
        )
        answers.profile = ("demo", "production", "headless", "custom")[idx]
    else:
        answers.profile = "demo"

    if answers.profile == "headless":
        answers.enable_headless = True
    elif getattr(args, "headless", False):
        if answers.profile == "demo":
            raise SystemExit(
                "Headless batch mode is not available with --profile demo (Express). "
                "Use --profile headless, --profile production --headless, "
                "or --profile custom without ECS Express."
            )
        answers.enable_headless = True
    elif interactive and answers.profile == "production":
        answers.enable_headless = ask_yes_no(
            "Enable headless batch-only deployment (S3 → Lambda → one-shot ECS, "
            "no always-on web UI)?",
            default=False,
        )

    if answers.profile == "custom" and interactive:
        answers.custom_overrides["USE_ECS_EXPRESS_MODE"] = (
            "True" if ask_yes_no("Use ECS Express Mode?", False) else "False"
        )
        answers.custom_overrides["USE_CLOUDFRONT"] = (
            "True" if ask_yes_no("Use CloudFront?", True) else "False"
        )
        answers.custom_overrides["RUN_USEAST_STACK"] = (
            "True"
            if answers.custom_overrides.get("USE_CLOUDFRONT") == "True"
            and ask_yes_no(
                "Deploy us-east-1 CloudFront stack (RUN_USEAST_STACK)?", True
            )
            else "False"
        )
        answers.custom_overrides["ENABLE_RESOURCE_DELETE_PROTECTION"] = (
            "True" if ask_yes_no("Enable delete protection?", True) else "False"
        )
        if getattr(args, "headless", False):
            if answers.custom_overrides.get("USE_ECS_EXPRESS_MODE") == "True":
                raise SystemExit(
                    "Headless batch mode requires USE_ECS_EXPRESS_MODE=False in "
                    "the custom profile."
                )
            answers.enable_headless = True
        elif profile_allows_headless_add_on(answers):
            answers.enable_headless = ask_yes_no(
                "Enable headless batch-only deployment (S3 → Lambda → one-shot ECS, "
                "no always-on web UI)?",
                default=False,
            )

    headless_err = headless_profile_error(answers)
    if headless_err:
        raise SystemExit(headless_err)

    try:
        account, region = get_aws_identity(args.region)
    except Exception as exc:
        raise SystemExit(f"AWS credentials error: {exc}") from exc

    answers.aws_account_id = args.account or account
    answers.aws_region = args.region or region

    if interactive and not args.account:
        answers.aws_account_id = ask("AWS account ID", answers.aws_account_id)
    if interactive and not args.region:
        answers.aws_region = ask("AWS region", answers.aws_region)

    default_prefix = f"{answers.profile.capitalize()}-Summarisation-"
    answers.cdk_prefix = args.cdk_prefix or (
        ask("CDK resource prefix (e.g. MyOrg-Summarisation-)", default_prefix)
        if interactive
        else default_prefix
    )
    if not answers.cdk_prefix.endswith("-"):
        answers.cdk_prefix += "-"

    if answers_use_headless(answers):
        answers.cognito_domain_prefix = ""
        print("Headless batch mode: skipping Cognito hosted UI domain (no web login).")

    resolve_globally_unique_install_names(
        answers,
        interactive=interactive,
        assume_yes=assume_yes,
        args=args,
    )

    # VPC
    if args.new_vpc_cidr:
        answers.vpc_mode = "new"
        answers.new_vpc_cidr = args.new_vpc_cidr
    elif args.vpc_name:
        answers.vpc_mode = "existing"
        answers.vpc_name = args.vpc_name
    elif interactive:
        vpc_idx = ask_choice(
            "VPC",
            ["Create new VPC", "Use existing VPC"],
        )
        answers.vpc_mode = "new" if vpc_idx == 0 else "existing"
    else:
        answers.vpc_mode = "existing"

    if answers.vpc_mode == "new":
        prompt_new_vpc_cidr(answers, interactive=interactive)
    else:
        if not answers.vpc_name:
            vpcs = list_vpcs(answers.aws_region)
            if not vpcs:
                raise SystemExit("No VPCs found in region.")
            if interactive:
                labels = [f"{v['name']} ({v['cidr']})" for v in vpcs]
                vidx = ask_choice("Select VPC", labels)
                answers.vpc_name = vpcs[vidx]["name"]
            else:
                raise SystemExit(
                    "--vpc-name required for existing VPC in non-interactive mode."
                )
        vpc_list = list_vpcs(answers.aws_region)
        vpc_id = next(
            (v["id"] for v in vpc_list if v["name"] == answers.vpc_name),
            None,
        )

        # Subnets (public only for Express or demo-style headless)
        apply_subnet_cli_flags(args, answers)
        public_subnets_only = answers_use_public_subnets_only(answers)
        if public_subnets_only:
            if interactive:
                answers.public_subnet_mode = ask_subnet_tier_mode("Public")
            else:
                answers.public_subnet_mode, _ = resolve_subnet_tier_modes(args)
            answers.private_subnet_mode = "auto"
            answers.private_subnet_names = []
            answers.private_subnet_cidrs = []
            answers.private_subnet_azs = []
        elif interactive:
            layout_idx = ask_choice(
                "Subnets in existing VPC",
                [
                    "Auto-discover public and private subnets",
                    "Use existing named subnets (both tiers)",
                    "Create new stack-specific subnets (both tiers)",
                    "Configure public and private separately",
                ],
                default_index=0,
            )
            if layout_idx < 3:
                shared_mode = SUBNET_TIER_MODES[layout_idx]
                answers.public_subnet_mode = shared_mode
                answers.private_subnet_mode = shared_mode
            else:
                answers.public_subnet_mode = ask_subnet_tier_mode("Public")
                answers.private_subnet_mode = ask_subnet_tier_mode("Private")
        else:
            answers.public_subnet_mode, answers.private_subnet_mode = (
                resolve_subnet_tier_modes(args)
            )

        azs = list_availability_zones(answers.aws_region)[:3]
        vpc_cidr = next(
            (v["cidr"] for v in vpc_list if v["name"] == answers.vpc_name),
            "",
        )
        if vpc_id:
            subnets = list_subnets_in_vpc(vpc_id, answers.aws_region)
            configure_subnet_tier(
                answers,
                "public",
                answers.public_subnet_mode,
                subnets,
                azs,
                interactive=interactive,
                vpc_cidr=vpc_cidr,
            )
            if not public_subnets_only:
                reserved_public = (
                    list(answers.public_subnet_cidrs)
                    if answers.public_subnet_mode == "create"
                    else []
                )
                configure_subnet_tier(
                    answers,
                    "private",
                    answers.private_subnet_mode,
                    subnets,
                    azs,
                    interactive=interactive,
                    vpc_cidr=vpc_cidr,
                    reserved_subnet_cidrs=reserved_public,
                )

        # Optional infra reuse
        if (
            vpc_id
            and interactive
            and ask_yes_no("Configure optional existing infra (IGW/ALB)?", False)
        ):
            igws = list_igws_for_vpc(vpc_id, answers.aws_region)
            if igws:
                # igws_ids = [g["id"] for g in igws]
                print("Internet gateways:", ", ".join(g["id"] for g in igws))
                answers.existing_igw_id = ask("EXISTING_IGW_ID (blank=new)", "")
            albs = list_albs_in_vpc(vpc_id, answers.aws_region)
            if albs:
                labels = [f"{a['name']} ({a['dns']})" for a in albs]
                if ask_yes_no("Reuse an existing ALB?", False):
                    aidx = ask_choice("Select ALB", labels)
                    answers.existing_alb_arn = albs[aidx]["arn"]
                    answers.existing_alb_dns = albs[aidx]["dns"]

    # Production TLS
    preset = merge_preset(answers_preset_profile(answers), answers.custom_overrides)
    use_express = preset.get("USE_ECS_EXPRESS_MODE") == "True"
    if (
        preset.get("USE_CLOUDFRONT") == "True"
        and preset.get("USE_ECS_EXPRESS_MODE") != "True"
        and not answers_use_headless(answers)
    ):
        if args.cert_arn:
            answers.acm_cert_arn = args.cert_arn
            answers.ssl_domain = args.domain or ""
        elif interactive:
            certs = list_acm_certificates(answers.aws_region)
            if not certs:
                raise SystemExit(f"No ISSUED ACM certificates in {answers.aws_region}.")
            labels = [f"{c['label']} — {c['arn']}" for c in certs]
            cidx = ask_choice("Select ACM certificate", labels)
            answers.acm_cert_arn = certs[cidx]["arn"]
            answers.ssl_domain = ask(
                "SSL certificate domain (Cognito/CloudFront URL)",
                certs[cidx]["domain"],
            )
        else:
            raise SystemExit(
                "--cert-arn and --domain required for production in non-interactive mode."
            )
        if interactive:
            answers.cloudfront_geo = ask(
                "CloudFront geo restriction (e.g. GB, blank=none)", ""
            )

    if not answers_use_headless(answers):
        warn_deprecated_pi_cli_flags(args)

    # Advanced add-ons (non-Pi agent)
    is_express = use_express and not answers_use_headless(answers)
    if answers_use_headless(answers) and interactive:
        mem = ask("ECS task memory (MB)", answers.ecs_memory)
        if mem:
            answers.ecs_memory = mem
        if getattr(args, "headless_output_notifications", False):
            answers.enable_headless_output_notifications = True
        elif getattr(args, "no_headless_output_notifications", False):
            answers.enable_headless_output_notifications = False
        elif ask_yes_no(
            "Enable email notifications when new analysis outputs are uploaded to S3?",
            default=True,
        ):
            answers.enable_headless_output_notifications = True
        if answers.enable_headless_output_notifications:
            default_email = getattr(args, "headless_notify_email", "") or ""
            while True:
                answers.headless_output_notify_email = ask(
                    "Notification email address (SNS subscription; confirm via AWS email)",
                    default_email,
                ).strip()
                email_error = validate_notify_email(
                    answers.headless_output_notify_email
                )
                if not email_error:
                    break
                print(email_error)
            default_iam_user = (
                getattr(args, "headless_output_iam_user", "").strip()
                or f"{answers.cdk_prefix}s3-output-reader"
            )
            iam_user = ask(
                "IAM user name for programmatic S3 output access",
                default_iam_user,
            ).strip()
            answers.headless_output_iam_user_name = iam_user or default_iam_user
    elif answers_use_headless(answers):
        if getattr(args, "headless_output_notifications", False):
            answers.enable_headless_output_notifications = True
            answers.headless_output_notify_email = (
                getattr(args, "headless_notify_email", "") or ""
            ).strip()
            answers.headless_output_iam_user_name = (
                getattr(args, "headless_output_iam_user", "").strip()
                or f"{answers.cdk_prefix}s3-output-reader"
            )
            email_error = validate_notify_email(answers.headless_output_notify_email)
            if email_error:
                raise SystemExit(email_error)
    elif interactive and ask_yes_no("Configure other optional add-ons?", False):
        if not is_express:
            if not answers.enable_service_connect:
                answers.enable_service_connect = ask_yes_no(
                    "Enable ECS Service Connect (without agent mode)?", False
                )
            answers.enable_s3_batch = ask_yes_no("Enable S3 batch ECS trigger?", False)
        mem = ask("ECS task memory (MB)", answers.ecs_memory)
        if mem:
            answers.ecs_memory = mem

    prompt_dynamo_usage_log_export_options(answers, args, interactive=interactive)
    prompt_bedrock_model_invocation_logging_options(
        answers, args, interactive=interactive
    )
    prompt_application_tagging_and_appregistry_options(
        answers, args, interactive=interactive
    )

    configure_app_config_options(answers, args, interactive=interactive)

    answers.python_path = args.python
    return answers


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------


def build_arg_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="Interactive CDK installer for llm_topic_modeller",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    p.add_argument(
        "--profile",
        choices=("demo", "production", "headless", "custom"),
        help="Base profile; use --profile headless or production --headless for batch-only",
    )
    p.add_argument(
        "--yes", action="store_true", help="Accept defaults / skip confirmations"
    )
    p.add_argument("--config-only", action="store_true", help="Write config files only")
    p.add_argument(
        "--deploy-only",
        action="store_true",
        help="Skip wizard; use existing cdk_config.env",
    )
    p.add_argument(
        "--synth-only", action="store_true", help="Run cdk synth only (no deploy)"
    )
    p.add_argument("--skip-deploy", action="store_true", help="Skip cdk deploy")
    p.add_argument(
        "--skip-quickstart",
        action="store_true",
        help="Skip post_cdk_build_quickstart.py",
    )
    p.add_argument(
        "--skip-stack-check",
        action="store_true",
        help="Do not check for existing CloudFormation stacks at startup",
    )
    p.add_argument(
        "--force-delete-stacks",
        action="store_true",
        help="If llm_topic_modeller stacks already exist, delete them before continuing "
        "(disables termination protection; implies consent in non-interactive mode)",
    )
    p.add_argument(
        "--skip-cdk-json", action="store_true", help="Do not update cdk.json"
    )
    p.add_argument(
        "--refresh-cdk-json", action="store_true", help="Force rewrite cdk.json app key"
    )
    p.add_argument("--python", help="Python executable for cdk.json")
    p.add_argument("--region", help="AWS region")
    p.add_argument("--account", help="AWS account ID")
    p.add_argument("--cdk-prefix", help="CDK resource prefix")
    p.add_argument("--cognito-prefix", help="Cognito user pool domain prefix")
    p.add_argument(
        "--s3-log-bucket",
        help="S3 log/config bucket name (globally unique; skips wizard suggestion)",
    )
    p.add_argument(
        "--s3-output-bucket",
        help="S3 output bucket name (globally unique; skips wizard suggestion)",
    )
    p.add_argument("--vpc-name", help="Existing VPC Name tag")
    p.add_argument("--new-vpc-cidr", help="CIDR for new VPC")
    p.add_argument(
        "--subnet-mode",
        choices=SUBNET_TIER_MODES,
        help="Subnet mode for both public and private tiers (overridden by per-tier flags)",
    )
    p.add_argument(
        "--public-subnet-mode",
        choices=SUBNET_TIER_MODES,
        help="Public subnet tier: auto, existing, or create",
    )
    p.add_argument(
        "--private-subnet-mode",
        choices=SUBNET_TIER_MODES,
        help="Private subnet tier: auto, existing, or create",
    )
    p.add_argument(
        "--public-subnet-names",
        help="Comma-separated existing public subnet names (with --public-subnet-mode existing)",
    )
    p.add_argument(
        "--private-subnet-names",
        help="Comma-separated existing private subnet names (with --private-subnet-mode existing)",
    )
    p.add_argument(
        "--headless",
        action="store_true",
        help="Batch-only add-on (production/custom without Express only; not valid with demo)",
    )
    p.add_argument(
        "--headless-output-notifications",
        action="store_true",
        help="Enable S3 output PutRequests alarm, SNS email, and IAM reader user (headless)",
    )
    p.add_argument(
        "--no-headless-output-notifications",
        action="store_true",
        help="Disable headless output notifications in interactive headless installs",
    )
    p.add_argument(
        "--headless-notify-email",
        help="Email for SNS notifications when headless outputs are uploaded",
    )
    p.add_argument(
        "--headless-output-iam-user",
        help="IAM user name for programmatic download of headless S3 outputs",
    )
    p.add_argument(
        "--dynamo-usage-log-export",
        action="store_true",
        help="Enable scheduled DynamoDB usage log export Lambda (EventBridge)",
    )
    p.add_argument(
        "--no-dynamo-usage-log-export",
        action="store_true",
        help="Disable DynamoDB usage log export in interactive installs",
    )
    p.add_argument(
        "--dynamo-export-schedule-time",
        metavar="HH:MM",
        help="UTC schedule time for DynamoDB usage log export (default: 06:00)",
    )
    p.add_argument(
        "--dynamo-export-schedule-days",
        choices=tuple(DYNAMO_EXPORT_SCHEDULE_DAY_PRESETS),
        help="Days to run DynamoDB usage log export (default: daily)",
    )
    p.add_argument(
        "--dynamo-export-s3-key",
        help="S3 object key for the exported CSV (default: reports/dynamodb-usage/...)",
    )
    p.add_argument(
        "--dynamo-export-date-attribute",
        help="DynamoDB attribute used for date filtering (default: timestamp)",
    )
    p.add_argument(
        "--bedrock-model-invocation-logging",
        action="store_true",
        help=(
            "Enable Bedrock model invocation logging to the S3 log/config bucket "
            "and CloudWatch Logs (account/Region-level)"
        ),
    )
    p.add_argument(
        "--no-bedrock-model-invocation-logging",
        action="store_true",
        help="Disable Bedrock model invocation logging in interactive installs",
    )
    p.add_argument(
        "--bedrock-invocation-s3-prefix",
        default="",
        help="S3 key prefix for Bedrock invocation logs (default: bedrock-logs)",
    )
    p.add_argument(
        "--bedrock-invocation-log-group",
        default="",
        help="CloudWatch log group for Bedrock invocation logs",
    )
    p.add_argument(
        "--bedrock-invocation-log-retention-days",
        type=int,
        default=None,
        metavar="DAYS",
        help="CloudWatch retention for Bedrock invocation logs (default: 90)",
    )
    p.add_argument(
        "--appregistry",
        action="store_true",
        help=(
            "Enable legacy AppRegistry / myApplications stack "
            "(opt-in; new customers cut off after 30 Jul 2026)"
        ),
    )
    p.add_argument(
        "--no-appregistry",
        action="store_true",
        help="Disable AppRegistry (default)",
    )
    p.add_argument(
        "--application-resource-group",
        action="store_true",
        help=(
            "Tag CDK-managed resources and create a tag-based AWS Resource Group "
            "(default on)"
        ),
    )
    p.add_argument(
        "--no-application-resource-group",
        action="store_true",
        help="Disable application Resource Group creation",
    )
    p.add_argument(
        "--application-name",
        default="",
        help="Application tag value / Resource Group filter (default: {CDK_PREFIX}llm-topic-modeller)",
    )
    p.add_argument(
        "--application-tag-key",
        default="",
        help="Application tag key applied to CDK-managed resources (default: Application)",
    )
    p.add_argument(
        "--create-headless-output-access-key",
        action="store_true",
        help="After deploy, create IAM access key for the headless output reader user",
    )
    p.add_argument(
        "--skip-headless-output-access-key",
        action="store_true",
        help="Do not offer to create IAM access key after headless deploy",
    )
    p.add_argument("--cert-arn", help="ACM certificate ARN (production)")
    p.add_argument("--domain", help="SSL certificate domain (production)")
    p.add_argument(
        "--skip-app-config-env",
        action="store_true",
        help="Do not write cdk/config/app_config.env from app_config.env.example",
    )
    p.add_argument(
        "--overwrite-app-config-env",
        action="store_true",
        help="Replace existing config/app_config.env from app_config.env.example (non-interactive)",
    )
    return p


def main(argv: Optional[Sequence[str]] = None) -> int:
    argv = list(argv) if argv is not None else None
    args = build_arg_parser().parse_args(argv)
    args._argv = tuple(argv if argv is not None else sys.argv[1:])
    CONFIG_DIR.mkdir(parents=True, exist_ok=True)

    print("llm_topic_modeller CDK installer\n")

    # Prerequisites
    try:
        cdk_version = check_cdk_cli()
        print(f"CDK CLI: {cdk_version}")
    except SystemExit:
        if not args.config_only:
            raise
        print("Warning: CDK CLI not found; config-only mode continuing.")

    if not args.config_only and not args.synth_only:
        try:
            _stack_account, stack_region = get_aws_identity(args.region)
            handle_existing_stacks_at_start(args, stack_region)
        except SystemExit:
            raise
        except Exception as exc:
            print(f"Existing stack check skipped: {exc}")

    python_exe: Optional[Path] = None
    if not args.skip_cdk_json:
        python_exe = resolve_python_executable(
            override=args.python,
            interactive=not args.yes and not args.deploy_only,
            assume_yes=args.yes or args.deploy_only,
        )
        write_cdk_json(
            python_exe,
            force=args.refresh_cdk_json or not CDK_JSON_PATH.is_file(),
            skip=args.skip_cdk_json,
        )
    elif CDK_JSON_PATH.is_file():
        # Best-effort parse python from existing cdk.json
        try:
            data = json.loads(CDK_JSON_PATH.read_text(encoding="utf-8"))
            app_cmd = data.get("app", "")
            if app_cmd:
                python_exe = Path(app_cmd.split()[0])
        except (json.JSONDecodeError, IndexError):
            pass

    values: Dict[str, str]
    if args.deploy_only:
        if not ENV_PATH.is_file():
            raise SystemExit(
                f"No config at {ENV_PATH}. Run without --deploy-only first."
            )
        values = read_env_file(ENV_PATH)
        pre_deploy_errors = validate_globally_unique_env_values(
            values
        ) + validate_new_vpc_cidr_env_values(values)
        if pre_deploy_errors:
            print("Pre-deploy configuration conflicts:")
            for err in pre_deploy_errors:
                print(f"  - {err}")
            return 1
        if not python_exe or not python_exe.is_file():
            python_exe = resolve_python_executable(
                override=args.python, interactive=False, assume_yes=True
            )
    else:
        answers = run_wizard(args)
        errors = validate_install_answers(answers)
        errors.extend(validate_subnet_answers(answers))
        errors.extend(enrich_existing_subnet_details_from_aws(answers))
        values = build_env_values(answers)
        errors.extend(validate_env_values(values))
        if errors:
            print("Configuration errors:")
            for err in errors:
                print(f"  - {err}")
            return 1

        if not python_exe:
            python_exe = resolve_python_executable(
                override=answers.python_path or args.python,
                interactive=not args.yes,
                assume_yes=args.yes,
            )
            if not args.skip_cdk_json:
                write_cdk_json(python_exe, force=args.refresh_cdk_json)

        print_summary(values, python_exe)
        if not args.yes and not ask_yes_no("Write config/cdk_config.env?", True):
            print("Aborted.")
            return 0

        write_env_file(ENV_PATH, values)
        if answers.write_app_config_env:
            write_app_config_env_file(
                answers,
                values,
                overwrite=answers.overwrite_app_config_env,
            )

    apply_cdk_runtime_env(values)

    if args.config_only:
        print("Config written. Exiting (--config-only).")
        return 0

    run_smoke_test_if_needed(python_exe, args)

    # Bootstrap prompt
    account = values.get("AWS_ACCOUNT_ID", "")
    region = values.get("AWS_REGION", "")
    if account and region:
        try:
            if cdk_bootstrap_needed(account, region):
                print(f"CDK bootstrap not found in {region}.")
                if args.yes or ask_yes_no(
                    f"Run cdk bootstrap aws://{account}/{region}?", True
                ):
                    run_cdk_command(
                        ["bootstrap", f"aws://{account}/{region}"],
                        check=True,
                    )
        except Exception as exc:
            print(f"Bootstrap check skipped: {exc}")

    verify_node_for_jsii()
    run_cdk_command(["synth"], check=True)

    if args.synth_only or args.skip_deploy:
        print("Synth complete. Skipping deploy.")
        return 0

    if not args.yes and not ask_yes_no("Run cdk deploy --all?", True):
        print("Deploy skipped.")
        return 0

    run_cdk_command(["deploy", "--all", "--require-approval", "broadening"], check=True)

    values = read_env_file(ENV_PATH)
    apply_post_deploy_fixup(values, assume_yes=args.yes)

    run_qs = False
    if args.skip_quickstart:
        print("Skipping post-deploy quickstart.")
    else:
        is_headless = values.get("ENABLE_HEADLESS_DEPLOYMENT") == "True"
        is_demo = values.get("USE_ECS_EXPRESS_MODE") == "True"
        if args.yes:
            run_qs = True
        elif is_headless:
            run_qs = ask_yes_no(
                "Run post_cdk_build_quickstart.py (CodeBuild image only; no ECS service)?",
                default=True,
            )
        else:
            run_qs = ask_yes_no(
                "Run post_cdk_build_quickstart.py (CodeBuild + scale ECS)?",
                default=is_demo,
            )
        if run_qs:
            if not python_exe:
                python_exe = resolve_python_executable(
                    assume_yes=True, interactive=False
                )
            run_quickstart(python_exe)

    values = read_env_file(ENV_PATH)
    is_headless = values.get("ENABLE_HEADLESS_DEPLOYMENT") == "True"
    if values.get("USE_ECS_EXPRESS_MODE") == "True" and not is_headless:
        from cdk_post_deploy import print_express_mode_next_steps

        if args.skip_quickstart or not run_qs:
            print_express_mode_next_steps(values)
        return 0

    if is_headless:
        from cdk_post_deploy import (
            print_headless_deployment_next_steps,
            print_headless_output_notification_steps,
            provision_headless_output_reader_access_key,
        )

        if args.skip_quickstart or not run_qs:
            print_headless_deployment_next_steps(values)
        if values.get("ENABLE_HEADLESS_OUTPUT_NOTIFICATIONS") == "True":
            print_headless_output_notification_steps(values)
            create_key = False
            if args.skip_headless_output_access_key:
                create_key = False
            elif args.create_headless_output_access_key:
                create_key = True
            elif not args.yes and not args.deploy_only:
                create_key = ask_yes_no(
                    "Create IAM access key for the headless output reader user now?",
                    default=True,
                )
            elif args.yes:
                create_key = True
            if create_key:
                provision_headless_output_reader_access_key(values)
        return 0

    print("\nDone. Next steps:")
    if values.get("USE_CLOUDFRONT") == "True":
        domain = values.get("SSL_CERTIFICATE_DOMAIN", "")
        cf = values.get("CLOUDFRONT_DOMAIN", "")
        if domain and cf and cf != "cloudfront_placeholder.net":
            print(f"  - Point DNS CNAME {domain} -> {cf}")
    elif values.get("USE_ECS_EXPRESS_MODE") == "True":
        ep = values.get("ECS_EXPRESS_COGNITO_REDIRECT_BASE", "")
        if ep:
            from cdk_config import normalize_https_redirect_url

            print(f"  - Express endpoint: {normalize_https_redirect_url(ep)}")
    if APP_CONFIG_ENV_PATH.is_file():
        print(f"  - App runtime config: {APP_CONFIG_ENV_PATH} (uploaded by quickstart)")
    print(f"  - Config: {ENV_PATH}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())