cdk/cdk_install.py

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

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),
and post_cdk_build_quickstart.py. When CloudFront is enabled, a one-off refresh deploy
of the regional stack runs after the domain is discovered so the response-headers policy
(CSP/CORS) and app config use the real *.cloudfront.net domain instead of the placeholder.

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 redaction.example.com --yes

    # Initial deploy using existing config, skip quickstart
    # (cdk_install performs initial deploys only; if a stack already exists it
    # aborts — remove the stack yourself first, or run post_cdk_build_quickstart.py
    # for post-deploy tasks that don't need a stack change)
    python cdk_install.py --deploy-only --skip-quickstart

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

    # Demo with agentic redaction (AgentCore CDK two-phase deploy is automatic)
    python cdk_install.py --profile demo --enable-agentic --yes

    # AgentCore two-phase deploy, phase 2 only: create the runtime once its ECR
    # image exists (CodeBuild success or manual push). Uses the AgentCore API
    # (boto3) — no cdk deploy / no stack update.
    python cdk_install.py --complete-agentcore

    # Demo with AgentCore URL (manual CLI deploy — see agent-redact/agentcore/README.md)
    python cdk_install.py --profile demo --enable-agentic --agent-orchestrator agentcore \\
        --agentcore-runtime-url https://your-runtime.example --yes --config-only

    # Fallback: Pi coding agent orchestrator inside the Express container
    python cdk_install.py --profile demo --enable-agentic --agent-orchestrator pi --yes --config-only

    # 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

    # Production with agentic redaction on dedicated hostname (legacy Fargate)
    python cdk_install.py --profile production --enable-agentic-legacy \\
        --agentic-alb-routing host --agentic-host-header agent.example.com --yes
"""

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"
# Agent runtime config file (renamed from pi_agent.env -> agent.env).
PI_AGENT_ENV_PATH = CONFIG_DIR / "agent.env"
PI_AGENT_ENV_EXAMPLE = REPO_ROOT / "config" / "agent.env.example"
# Legacy path kept so an existing pi_agent.env is detected/migrated if present.
PI_AGENT_ENV_LEGACY_PATH = CONFIG_DIR / "pi_agent.env"
AGENTIC_ALB_ROUTING_MODES = ("path", "host", "both")
AGENT_ORCHESTRATOR_CHOICES = ("pi", "langgraph", "agentcore")

DEFAULT_POLICY_FILE_LOCATIONS = (
    "policies/textract_policy.json",
    "policies/comprehend_policy.json",
)
PI_AGENTCORE_INVOKE_POLICY_FILE = "policies/pi_agentcore_invoke_policy.json"

# Bedrock models selectable for the AgentCore runtime engine. Mirrors
# BEDROCK_MODELS in agent-redact/pi/pi_agent_config.py (kept local to avoid
# importing the heavy Pi agent config from the installer). The runtime's model
# is fixed at creation via the AGENT_DEFAULT_MODEL env var; changing it later
# requires recreating the runtime.
AGENTCORE_DEFAULT_BEDROCK_MODEL = "anthropic.claude-sonnet-4-6"
AGENTCORE_BEDROCK_MODEL_CHOICES = (
    ("anthropic.claude-sonnet-4-6", "Anthropic Claude Sonnet 4.6"),
    ("amazon.nova-pro-v1:0", "Amazon Nova Pro"),
    ("nvidia.nemotron-super-3-120b", "NVIDIA Nemotron Super 3 120B"),
    ("mistral.devstral-2-123b", "Mistral Devstral 2 123B"),
)

AGENTCORE_CDK_DEPLOY_SUMMARY = (
    "AgentCore CDK deploy: CodeBuild pushes the runtime image to ECR (phase 1), "
    "then the installer creates the Bedrock runtime and sets AGENTCORE_RUNTIME_URL "
    "automatically (phase 2)."
)

AGENTCORE_MANUAL_DEPLOY_CHECKLIST = """
AgentCore manual CLI deploy:
  1. Package runtime: python agent-redact/agentcore/package_runtime.py \\
       --target <AgentCoreProject>/app/RedactionAgent
     (Windows/OneDrive: set UV_LINK_MODE=copy before agentcore deploy)
  2. Deploy runtime: cd <AgentCoreProject> && agentcore deploy
  3. Copy invocationUrl from `agentcore status` (base URL, no /invocations suffix)
  4. Re-run cdk_install.py --config-only with --agentcore-runtime-url <URL>
     or update config/agent.env + upload to S3, then restart the agentic Express service
""".strip()

# Kept for backwards compatibility (tests / docs may reference the name).
AGENTCORE_DEPLOY_CHECKLIST = (
    f"{AGENTCORE_CDK_DEPLOY_SUMMARY}\n\n{AGENTCORE_MANUAL_DEPLOY_CHECKLIST}"
)
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 = "RedactionStack"
CLOUDFRONT_STACK = "RedactionStackCloudfront"
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": "True",
    "CLOUDFRONT_AUTH_MODE": "magic-link",
    "ENABLE_CLOUDFRONT_WAF": "False",
    # Lock the Express managed ALB SG(s) to CloudFront after deploy (boto3 quickstart step).
    "RESTRICT_ALB_INGRESS_TO_CLOUDFRONT": "True",
    "RUN_USEAST_STACK": "False",
    "ENABLE_RESOURCE_DELETE_PROTECTION": "False",
    # Default OFF: AWS is retiring creation of new AppRegistry Applications.
    "ENABLE_APPREGISTRY": "False",
    "ACM_SSL_CERTIFICATE_ARN": "",
    "SSL_CERTIFICATE_DOMAIN": "",
}

PRODUCTION_PRESET: Dict[str, str] = {
    "USE_ECS_EXPRESS_MODE": "False",
    "USE_CLOUDFRONT": "True",
    "CLOUDFRONT_AUTH_MODE": "cognito",
    "ENABLE_CLOUDFRONT_WAF": "False",
    "RUN_USEAST_STACK": "False",
    "ENABLE_RESOURCE_DELETE_PROTECTION": "True",
    # Default OFF: AWS is retiring creation of new AppRegistry Applications.
    "ENABLE_APPREGISTRY": "False",
}

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",
    # Default OFF: AWS is retiring creation of new AppRegistry Applications.
    "ENABLE_APPREGISTRY": "False",
    "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 = ""
    s3_malware_scan_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_agentic_express: bool = False
    enable_agentic_legacy: bool = False
    enable_service_connect: bool = False
    enable_s3_batch: bool = False
    enable_headless: bool = False
    ecs_memory: str = "8192"
    ecs_cpu: str = "2048"
    agentic_alb_routing: str = "path"
    agentic_alb_path_prefix: str = "/agent"
    agentic_alb_host_header: str = ""
    agentic_alb_listener_rule_priority: str = ""
    agentic_gradio_port: str = "7862"
    sc_discovery_name: str = "redaction"
    pi_default_provider: str = "amazon-bedrock"
    agent_orchestrator: str = "pi"
    enable_agentcore_runtime: bool = False
    enable_agentcore_cdk_deploy: bool = False
    enable_agentcore_cdk_runtime: bool = False
    agentcore_runtime_url: str = ""
    agentcore_api_key: str = ""
    agentcore_bedrock_model: str = ""
    allow_empty_agentcore_url: bool = False
    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 agentic_enabled(self) -> bool:
        return self.enable_agentic_express or self.enable_agentic_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 "redaction-app"
    prefix = prefix.rstrip("-")
    if not prefix:
        return "redaction-app"
    for reserved in COGNITO_DOMAIN_RESERVED_SUBSTRINGS:
        if reserved in prefix:
            return "redaction-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_agentic_path_prefix(raw: str) -> str:
    segment = (raw or "agent").strip().strip("/")
    return f"/{segment}" if segment else "/agent"


def default_agentic_listener_priority(use_cloudfront: bool = False) -> str:
    """Default agentic ALB 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}AgentExpressService",
        "ECS_PI_SERVICE_NAME": f"{prefix}AgentService",
        "COGNITO_USER_POOL_CLIENT_SECRET_NAME": f"{prefix}ParamCognitoSecret",
        "CODEBUILD_PROJECT_NAME": f"{prefix}CodeBuildProject",
        "CODEBUILD_PI_PROJECT_NAME": f"{prefix}CodeBuildAgentProject",
    }


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_MALWARE_SCAN_BUCKET_NAME": f"{prefix}s3-malware-scan",
    }


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,
    )
    answers.s3_malware_scan_bucket_name = _prompt_globally_unique_s3_bucket(
        "S3 malware scan staging bucket",
        defaults.get("S3_MALWARE_SCAN_BUCKET_NAME", ""),
        account_id,
        interactive=interactive,
        assume_yes=assume_yes,
        cli_override=getattr(args, "s3_malware_scan_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",
        "S3_MALWARE_SCAN_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 normalize_agent_orchestrator(raw: str) -> str:
    value = (raw or "pi").strip().lower()
    if value in AGENT_ORCHESTRATOR_CHOICES:
        return value
    return "pi"


def normalize_agentcore_runtime_url(raw: str) -> str:
    """Base runtime URL only — strip trailing slash and accidental /invocations suffix."""
    from cdk_functions import normalize_agentcore_runtime_url as _normalize

    return _normalize(raw)


def normalize_agentcore_bedrock_model(raw: str) -> str:
    """Return a trimmed Bedrock model id, defaulting to Sonnet 4.6 when empty."""
    model = (raw or "").strip()
    return model or AGENTCORE_DEFAULT_BEDROCK_MODEL


def validate_bedrock_model_access(
    model_id: str, region: str
) -> Tuple[Optional[bool], str]:
    """
    Best-effort check that the calling AWS identity can use *model_id* in *region*.

    Returns ``(status, message)`` where ``status`` is ``True`` (access confirmed),
    ``False`` (definitively no access) or ``None`` (could not determine, e.g. missing
    boto3, throttling, or an ambiguous error). Never raises.

    Strategy: query the Bedrock control-plane for foundation-model availability
    (entitlement + authorization) first; if that is inconclusive, fall back to a
    minimal ``bedrock-runtime`` converse probe (1 token) that exercises the exact
    model string the runtime will use.
    """
    model_id = (model_id or "").strip()
    if not model_id:
        return None, "No Bedrock model id provided."
    try:
        import boto3  # noqa: PLC0415
        from botocore.exceptions import (  # noqa: PLC0415
            BotoCoreError,
            ClientError,
        )
    except Exception as exc:  # noqa: BLE001
        return None, f"boto3 unavailable, skipping Bedrock access check ({exc})."

    # 1) Control-plane entitlement/authorization (does not consume tokens).
    try:
        bedrock = boto3.client("bedrock", region_name=region)
        avail = bedrock.get_foundation_model_availability(modelId=model_id)
        entitlement = (avail.get("entitlementAvailability") or "").upper()
        authorization = (avail.get("authorizationStatus") or "").upper()
        region_av = (avail.get("regionAvailability") or "").upper()
        if region_av and region_av != "AVAILABLE":
            return (
                False,
                f"Model '{model_id}' is not available in region {region} "
                f"(regionAvailability={region_av}).",
            )
        if entitlement == "AVAILABLE" and authorization in ("", "AUTHORIZED"):
            return True, f"Bedrock access confirmed for '{model_id}' in {region}."
        if entitlement == "NOT_AVAILABLE":
            return (
                False,
                f"Model access for '{model_id}' is not granted in this account. "
                "Request access in the Bedrock console (Model access), then retry.",
            )
        # Otherwise fall through to the runtime probe for a definitive answer.
    except ClientError as exc:
        code = exc.response.get("Error", {}).get("Code", "")
        if code in ("AccessDeniedException", "UnrecognizedClientException"):
            return (
                None,
                "Could not check model availability (missing bedrock:GetFoundation"
                "ModelAvailability permission); will try a runtime probe instead.",
            )
        # ValidationException often means the id is an inference-profile id the
        # control plane doesn't accept — fall through to the runtime probe.
    except (BotoCoreError, Exception):  # noqa: BLE001
        pass

    # 2) Runtime converse probe (consumes ~1 output token).
    try:
        rt = boto3.client("bedrock-runtime", region_name=region)
        rt.converse(
            modelId=model_id,
            messages=[{"role": "user", "content": [{"text": "ping"}]}],
            inferenceConfig={"maxTokens": 1},
        )
        return True, f"Bedrock access confirmed for '{model_id}' in {region}."
    except ClientError as exc:
        code = exc.response.get("Error", {}).get("Code", "")
        msg = exc.response.get("Error", {}).get("Message", str(exc))
        if code in ("AccessDeniedException",):
            return (
                False,
                f"Access denied invoking '{model_id}' in {region}. Grant model "
                "access in the Bedrock console and ensure the caller has "
                "bedrock:InvokeModel / bedrock:Converse permissions.",
            )
        if code in ("ResourceNotFoundException",):
            return (
                False,
                f"Model '{model_id}' was not found in {region}. Check the model id "
                "and region, or use an inference-profile id (e.g. 'us.<model>').",
            )
        if code in ("ValidationException",) and "inference profile" in msg.lower():
            return (
                None,
                f"Model '{model_id}' requires an inference profile id in {region} "
                "(e.g. prefix with 'us.'). Access could not be confirmed automatically.",
            )
        return None, f"Could not confirm access to '{model_id}' ({code or exc})."
    except (BotoCoreError, Exception) as exc:  # noqa: BLE001
        return None, f"Could not confirm access to '{model_id}' ({exc})."


def report_bedrock_model_access(
    model_id: str,
    region: str,
    *,
    interactive: bool,
    assume_yes: bool,
) -> bool:
    """
    Run :func:`validate_bedrock_model_access` and surface the result.

    Returns ``True`` to proceed, ``False`` to abort. Confirmed access proceeds
    silently-ish; a definitive "no access" prompts the user (interactive) or warns
    loudly and proceeds (non-interactive). Inconclusive results always proceed.
    """
    status, message = validate_bedrock_model_access(model_id, region)
    if status is True:
        print(f"  [ok] {message}")
        return True
    if status is None:
        print(f"  [warn] {message}")
        return True
    # Definitive no-access.
    print(f"\n  [ERROR] {message}")
    if interactive and not assume_yes:
        return ask_yes_no(
            "Continue anyway with this model (the AgentCore runtime will fail "
            "until access is granted)?",
            default=False,
        )
    print(
        "  Proceeding anyway (non-interactive). Grant Bedrock model access before "
        "invoking the AgentCore runtime, or re-run with a different "
        "--agentcore-model."
    )
    return True


def finalize_agentcore_bedrock_model(
    answers: InstallAnswers,
    args: argparse.Namespace,
    *,
    interactive: bool,
    assume_yes: bool,
) -> None:
    """
    Resolve the AgentCore Bedrock model and (optionally) validate account access.

    Interactive runs prompt for a model (default Sonnet 4.6); non-interactive runs
    keep any CLI/answers value or fall back to the default. Access is validated for
    the calling AWS identity unless ``--skip-bedrock-model-check`` was passed.
    """
    if interactive and not assume_yes and not getattr(args, "agentcore_model", None):
        print("\n--- AgentCore engine model (Bedrock) ---")
        print(
            "The AgentCore runtime's model is fixed when the runtime is created and "
            "cannot be changed from the Gradio UI later."
        )
        labels = [
            f"{label} ({model})" for model, label in AGENTCORE_BEDROCK_MODEL_CHOICES
        ]
        labels.append("Other (enter a custom Bedrock model id)")
        current = normalize_agentcore_bedrock_model(answers.agentcore_bedrock_model)
        default_idx = next(
            (
                i
                for i, (model, _) in enumerate(AGENTCORE_BEDROCK_MODEL_CHOICES)
                if model == current
            ),
            0,
        )
        idx = ask_choice(
            "Which Bedrock model should the AgentCore engine use?",
            labels,
            default_index=default_idx,
        )
        if idx < len(AGENTCORE_BEDROCK_MODEL_CHOICES):
            answers.agentcore_bedrock_model = AGENTCORE_BEDROCK_MODEL_CHOICES[idx][0]
        else:
            answers.agentcore_bedrock_model = normalize_agentcore_bedrock_model(
                ask("Bedrock model id", current)
            )
    else:
        answers.agentcore_bedrock_model = normalize_agentcore_bedrock_model(
            answers.agentcore_bedrock_model
        )

    model = answers.agentcore_bedrock_model
    if getattr(args, "skip_bedrock_model_check", False):
        print(f"  AgentCore Bedrock model: {model} (access check skipped)")
        return
    region = (answers.aws_region or "").strip() or _default_region_for_check()
    print(f"  Checking Bedrock access for '{model}' in {region} ...")
    proceed = report_bedrock_model_access(
        model, region, interactive=interactive, assume_yes=assume_yes
    )
    if not proceed:
        raise SystemExit(
            "Aborted: no access to the selected AgentCore Bedrock model. "
            "Grant model access in the Bedrock console or choose another model."
        )


def _default_region_for_check() -> str:
    """Resolve a region for the Bedrock access probe (answers/env/AWS default)."""
    for key in ("AWS_REGION", "AWS_DEFAULT_REGION"):
        val = (os.environ.get(key) or "").strip()
        if val:
            return val
    try:
        import boto3  # noqa: PLC0415

        region = boto3.session.Session().region_name
        if region:
            return str(region)
    except Exception:  # noqa: BLE001
        pass
    return "us-east-1"


def default_agent_orchestrator_for_answers(answers: "InstallAnswers") -> str:
    """Demo Express agent mode defaults to AgentCore; production/legacy stays on Pi."""
    if answers.profile == "demo" and answers.enable_agentic_express:
        return "agentcore"
    return "pi"


def apply_demo_agentcore_orchestrator_defaults(
    answers: "InstallAnswers", args: argparse.Namespace
) -> None:
    """When demo + Express agent mode without an explicit orchestrator, use AgentCore."""
    if answers.profile != "demo" or not answers.enable_agentic_express:
        return
    if getattr(args, "agent_orchestrator", None):
        return
    answers.agent_orchestrator = "agentcore"
    answers.enable_agentcore_runtime = True
    if not getattr(args, "agentcore_runtime_url", None):
        answers.enable_agentcore_cdk_deploy = True


def merge_policy_file_locations(
    *,
    agentcore_enabled: bool,
    existing_raw: str = "",
) -> str:
    """ECS task-role JSON policy paths written to POLICY_FILE_LOCATIONS in cdk_config.env."""
    try:
        from cdk_config import parse_comma_separated_list
    except ImportError:
        parse_comma_separated_list = lambda value: [  # noqa: E731
            part.strip() for part in (value or "").split(",") if part.strip()
        ]

    policies: List[str] = []
    if existing_raw.strip():
        policies = list(parse_comma_separated_list(existing_raw))
    if not policies:
        policies = list(DEFAULT_POLICY_FILE_LOCATIONS)
    if agentcore_enabled and PI_AGENTCORE_INVOKE_POLICY_FILE not in policies:
        policies.append(PI_AGENTCORE_INVOKE_POLICY_FILE)
    return json.dumps(policies)


def validate_install_answers(answers: "InstallAnswers") -> List[str]:
    errors: List[str] = []
    msg = headless_profile_error(answers)
    if msg:
        errors.append(msg)
    orchestrator = normalize_agent_orchestrator(answers.agent_orchestrator)
    if orchestrator not in AGENT_ORCHESTRATOR_CHOICES:
        errors.append(
            "AGENT_ORCHESTRATOR must be one of "
            f"{list(AGENT_ORCHESTRATOR_CHOICES)}; got '{answers.agent_orchestrator}'."
        )
    if answers.agentic_enabled and orchestrator == "agentcore":
        if not (answers.agentcore_runtime_url or "").strip():
            if not answers.allow_empty_agentcore_url and not (
                answers.enable_agentcore_cdk_deploy
                or answers.enable_agentcore_cdk_runtime
            ):
                errors.append(
                    "AGENTCORE_RUNTIME_URL is required when AGENT_ORCHESTRATOR=agentcore. "
                    "Deploy the runtime first (see agent-redact/agentcore/README.md), pass "
                    "--agentcore-runtime-url, or use --enable-agentcore-cdk-deploy for the "
                    "CDK-native ECR path. Non-interactive installs must include the URL or "
                    "enable the CDK deploy flags."
                )
    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
    if answers.s3_malware_scan_bucket_name:
        s3_bucket_names["S3_MALWARE_SCAN_BUCKET_NAME"] = (
            answers.s3_malware_scan_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_CPU_SIZE": answers.ecs_cpu,
            "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_agentic_express else "False"
            ),
            "ENABLE_PI_AGENT_ECS_SERVICE": (
                "True" if answers.enable_agentic_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",
            }
        )

    use_cloudfront = values.get("USE_CLOUDFRONT") == "True"
    if answers.agentic_enabled:
        orchestrator = normalize_agent_orchestrator(answers.agent_orchestrator)
        agentcore_enabled = (
            answers.enable_agentcore_runtime or orchestrator == "agentcore"
        )
        values.update(
            {
                "AGENT_GRADIO_PORT": answers.agentic_gradio_port,
                "ECS_SERVICE_CONNECT_DISCOVERY_NAME": answers.sc_discovery_name,
                "AGENT_ORCHESTRATOR": orchestrator,
                "ENABLE_AGENTCORE_RUNTIME": "True" if agentcore_enabled else "False",
            }
        )
        if answers.agentcore_runtime_url.strip():
            values["AGENTCORE_RUNTIME_URL"] = normalize_agentcore_runtime_url(
                answers.agentcore_runtime_url
            )
        if answers.agentcore_api_key.strip():
            values["AGENTCORE_API_KEY"] = answers.agentcore_api_key.strip()
        if agentcore_enabled:
            values["AGENTCORE_BEDROCK_MODEL"] = normalize_agentcore_bedrock_model(
                answers.agentcore_bedrock_model
            )
        if answers.enable_agentcore_cdk_deploy:
            values["AGENTCORE_CDK_DEPLOY"] = "True"
            values["ENABLE_AGENTCORE_RUNTIME"] = "True"
        if answers.enable_agentcore_cdk_runtime:
            values["ENABLE_AGENTCORE_CDK_RUNTIME"] = "True"
            values["AGENTCORE_CDK_DEPLOY"] = "True"
            values["ENABLE_AGENTCORE_RUNTIME"] = "True"
        existing_policy_raw = ""
        if ENV_PATH.is_file():
            existing_policy_raw = (
                read_env_file(ENV_PATH).get("POLICY_FILE_LOCATIONS") or ""
            )
        values["POLICY_FILE_LOCATIONS"] = merge_policy_file_locations(
            agentcore_enabled=agentcore_enabled,
            existing_raw=existing_policy_raw,
        )
        if answers.enable_agentic_express:
            values["ECS_EXPRESS_SC_PORT_NAME"] = "port-7860"
            values["ECS_PI_EXPRESS_SC_PORT_NAME"] = (
                f"port-{answers.agentic_gradio_port}"
            )
        else:
            routing = answers.agentic_alb_routing.strip().lower()
            path_prefix = normalize_agentic_path_prefix(answers.agentic_alb_path_prefix)
            priority = (
                answers.agentic_alb_listener_rule_priority.strip()
                or default_agentic_listener_priority(use_cloudfront)
            )
            values.update(
                {
                    "AGENT_ALB_ROUTING": routing,
                    "AGENT_ALB_PATH_PREFIX": path_prefix,
                    "AGENT_ALB_LISTENER_RULE_PRIORITY": priority,
                }
            )
            if routing in ("host", "both"):
                values["AGENT_ALB_HOST_HEADER"] = (
                    answers.agentic_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
            )

    if values.get("ENABLE_APPREGISTRY") == "True":
        values["APPREGISTRY_STACK_NAME"] = (
            f"{answers.cdk_prefix}{APPREGISTRY_STACK_SUFFIX}"
        )

    return values


def validate_env_values(
    values: Dict[str, str], *, allow_empty_agentcore_url: bool = False
) -> 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."
            )

    agentic_ecs = values.get("ENABLE_PI_AGENT_ECS_SERVICE") == "True"
    agentic_express = values.get("ENABLE_PI_AGENT_EXPRESS_SERVICE") == "True"
    if agentic_ecs and agentic_express:
        errors.append("Enable at most one agentic deployment mode (ECS or Express).")
    if agentic_express and not express:
        errors.append(
            "ENABLE_PI_AGENT_EXPRESS_SERVICE requires USE_ECS_EXPRESS_MODE=True."
        )
    if agentic_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"),
        ("S3_MALWARE_SCAN_BUCKET_NAME", "s3-malware-scan"),
    ):
        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 agentic_ecs and values.get("ENABLE_ECS_SERVICE_CONNECT") != "True":
        errors.append(
            "ENABLE_PI_AGENT_ECS_SERVICE=True requires ENABLE_ECS_SERVICE_CONNECT=True."
        )

    orchestrator = normalize_agent_orchestrator(values.get("AGENT_ORCHESTRATOR", "pi"))
    if orchestrator not in AGENT_ORCHESTRATOR_CHOICES:
        errors.append(
            f"AGENT_ORCHESTRATOR must be one of {list(AGENT_ORCHESTRATOR_CHOICES)}; "
            f"got '{values.get('AGENT_ORCHESTRATOR')}'."
        )
    agentcore_enabled = values.get("ENABLE_AGENTCORE_RUNTIME") == "True"
    if (agentic_ecs or agentic_express) and orchestrator == "agentcore":
        if not (values.get("AGENTCORE_RUNTIME_URL") or "").strip():
            cdk_deploy = values.get("AGENTCORE_CDK_DEPLOY") == "True"
            if not allow_empty_agentcore_url and not cdk_deploy:
                errors.append(
                    "AGENTCORE_RUNTIME_URL is required when AGENT_ORCHESTRATOR=agentcore "
                    "unless AGENTCORE_CDK_DEPLOY=True (CDK-native ECR/runtime path)."
                )
    if agentcore_enabled and orchestrator != "agentcore":
        errors.append(
            "ENABLE_AGENTCORE_RUNTIME=True requires AGENT_ORCHESTRATOR=agentcore."
        )

    if headless and (
        agentic_ecs
        or agentic_express
        or values.get("ENABLE_ECS_SERVICE_CONNECT") == "True"
    ):
        errors.append(
            "ENABLE_HEADLESS_DEPLOYMENT is incompatible with agentic redaction or Service Connect."
        )

    agentic_routing = (values.get("AGENT_ALB_ROUTING") or "").strip().lower()
    if agentic_ecs:
        if agentic_routing not in AGENTIC_ALB_ROUTING_MODES:
            errors.append(
                f"AGENT_ALB_ROUTING must be one of {list(AGENTIC_ALB_ROUTING_MODES)}; got '{agentic_routing}'."
            )
        elif (
            agentic_routing in ("host", "both")
            and not (values.get("AGENT_ALB_HOST_HEADER") or "").strip()
        ):
            errors.append(
                "AGENT_ALB_HOST_HEADER is required when AGENT_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"
    agentic_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}"

    return {
        "COGNITO_AUTH": "False" if headless or agentic_express else "True",
        "RUN_AWS_FUNCTIONS": "True",
        "DISPLAY_FILE_NAMES_IN_LOGS": "False",
        "SESSION_OUTPUT_FOLDER": "True",
        "SAVE_LOGS_TO_DYNAMODB": "True",
        "SHOW_COSTS": "True",
        "SHOW_WHOLE_DOCUMENT_TEXTRACT_CALL_OPTIONS": "True",
        "LOAD_PREVIOUS_TEXTRACT_JOBS_S3": "True",
        "RUN_ALL_EXAMPLES_THROUGH_AWS": "True",
        "SHOW_EXAMPLES": "True",
        "CUSTOM_VLM_BACKEND": "bedrock_vlm",
        "INCLUDE_FACE_IDENTIFICATION_TEXTRACT_OPTION": "True",
        "SHOW_SUMMARISATION": "True",
        "DOCUMENT_REDACTION_BUCKET": _name("S3_LOG_CONFIG_BUCKET_NAME", "s3-logs"),
        "TEXTRACT_WHOLE_DOCUMENT_ANALYSIS_BUCKET": _name(
            "S3_OUTPUT_BUCKET_NAME", "s3-output"
        ),
        "SCAN_UPLOADS_FOR_MALWARE": "True",
        "MALWARE_SCAN_S3_BUCKET": _name(
            "S3_MALWARE_SCAN_BUCKET_NAME", "s3-malware-scan"
        ),
        "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"
        ),
        "SAVE_OUTPUTS_TO_S3": "True" if headless else "False",
        "S3_OUTPUTS_BUCKET": (
            _name("S3_OUTPUT_BUCKET_NAME", "s3-output") if headless else ""
        ),
        "S3_OUTPUTS_FOLDER": _name("S3_OUTPUTS_FOLDER", "") if headless else "",
    }


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 resolve_doc_redaction_gradio_url(answers: InstallAnswers) -> str:
    """Backend URL for Pi / AgentCore runtime_config (Service Connect or main Express HTTPS)."""
    sc_url = f"http://{answers.sc_discovery_name}:7860"
    orchestrator = normalize_agent_orchestrator(answers.agent_orchestrator)
    if orchestrator != "agentcore":
        return sc_url
    endpoint = fetch_stack_output(
        REGIONAL_STACK, "ExpressServiceEndpoint", answers.aws_region
    )
    if endpoint:
        from cdk_config import normalize_https_redirect_url

        return normalize_https_redirect_url(endpoint)
    return sc_url


def build_pi_agent_env_values(answers: InstallAnswers) -> Dict[str, str]:
    """Runtime settings for the Agent Gradio app (uploaded to S3 as agent.env)."""
    values = {
        "AGENT_DEPLOYMENT_PROFILE": "aws-ecs",
        "AGENT_DEFAULT_PROVIDER": answers.pi_default_provider,
        "DOC_REDACTION_GRADIO_URL": resolve_doc_redaction_gradio_url(answers),
        "RUN_AWS_FUNCTIONS": "True",
        "AWS_REGION": answers.aws_region,
        "AGENT_GRADIO_PORT": answers.agentic_gradio_port,
        "AGENT_DEFAULT_OCR_METHOD": "AWS Textract service - all PDF types",
        "AGENT_DEFAULT_PII_METHOD": "AWS Comprehend",
    }
    if answers.enable_agentic_express:
        values["RUN_FASTAPI"] = "True"
    orchestrator = normalize_agent_orchestrator(answers.agent_orchestrator)
    values["AGENT_ORCHESTRATOR"] = orchestrator
    if orchestrator == "agentcore":
        # The AgentCore runtime uses Bedrock; pin the model so the container's
        # AGENT_DEFAULT_MODEL matches what the runtime was created with.
        values["AGENT_DEFAULT_PROVIDER"] = "amazon-bedrock"
        values["AGENT_DEFAULT_MODEL"] = normalize_agentcore_bedrock_model(
            answers.agentcore_bedrock_model
        )
    if answers.agentcore_runtime_url.strip():
        values["AGENTCORE_RUNTIME_URL"] = normalize_agentcore_runtime_url(
            answers.agentcore_runtime_url
        )
    if answers.agentcore_api_key.strip():
        values["AGENTCORE_API_KEY"] = answers.agentcore_api_key.strip()
    if answers.enable_agentic_express:
        return values
    path_prefix = normalize_agentic_path_prefix(answers.agentic_alb_path_prefix)
    if answers.agentic_alb_routing.strip().lower() in ("path", "both"):
        # The agent app only consumes AGENT_ROOT_PATH; ROOT_PATH / FASTAPI_ROOT_PATH
        # are the main redaction app's variables and must not be set here.
        values["AGENT_ROOT_PATH"] = path_prefix
    return values


def write_pi_agent_env_file(
    answers: InstallAnswers,
    *,
    overwrite: bool = False,
) -> Optional[Path]:
    if not answers.agentic_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)

    # Prefer agent.env; migrate an existing legacy pi_agent.env when agent.env is absent.
    existing_source = None
    if PI_AGENT_ENV_PATH.is_file():
        existing_source = PI_AGENT_ENV_PATH
    elif PI_AGENT_ENV_LEGACY_PATH.is_file():
        existing_source = PI_AGENT_ENV_LEGACY_PATH

    if existing_source is not None and not overwrite:
        existing = read_env_file(existing_source)
        existing.update(updates)
        backup_file(PI_AGENT_ENV_PATH) if PI_AGENT_ENV_PATH.is_file() else None
        lines = [f"{k}={v}" for k, v in existing.items()]
        PI_AGENT_ENV_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8")
        note = (
            " (migrated from pi_agent.env)"
            if existing_source is PI_AGENT_ENV_LEGACY_PATH
            else ""
        )
        print(f"Updated {PI_AGENT_ENV_PATH} (AWS agentic runtime keys merged){note}")
        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 — agentic runtime config for AWS ECS (agent.env)",
        *[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")
    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_RESOURCE_DELETE_PROTECTION",
        "ENABLE_APPREGISTRY",
        "VPC_NAME",
        "NEW_VPC_CIDR",
        "ACM_SSL_CERTIFICATE_ARN",
        "SSL_CERTIFICATE_DOMAIN",
        "ENABLE_PI_AGENT_EXPRESS_SERVICE",
        "ENABLE_PI_AGENT_ECS_SERVICE",
        "AGENT_ALB_ROUTING",
        "AGENT_ALB_PATH_PREFIX",
        "AGENT_ALB_HOST_HEADER",
    ]
    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 legacy us-east-1 CloudFront stack lookup (CloudFront is in RedactionStack)."""
    if not config_values:
        return False
    run_useast = config_values.get("RUN_USEAST_STACK")
    return run_useast == "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_doc_redaction_stacks(
    regional_region: str,
    config_values: Optional[Dict[str, str]] = None,
) -> List[ExistingStack]:
    """Find deployed doc_redaction 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 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_doc_redaction_stacks(
            regional_region, config_values
        )
    except Exception as exc:
        print(f"Existing stack check skipped: {exc}")
        return

    if not existing:
        return

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

    # cdk_install performs INITIAL deploys only. It never updates an existing
    # stack in place (repeated in-place updates caused imported-resource drift,
    # e.g. deleted CodeBuild roles/projects) and never interactively offers to
    # delete a stack. If the caller explicitly passes --force-delete-stacks we
    # honour it; otherwise abort with guidance so the operator can tear the
    # stack(s) down themselves and re-run for a clean initial deploy.
    if args.force_delete_stacks:
        force_delete_cloudformation_stacks(existing)
        print("Existing stacks deleted.\n")
        return

    raise SystemExit(
        "\ncdk_install only performs initial deploys; it will not update the "
        "existing stack(s) above in place. To deploy cleanly, remove the "
        "stack(s) yourself (CloudFormation console or your own tooling) and "
        "re-run cdk_install.py. For post-deploy tasks that don't require a "
        "stack change (CodeBuild images, ECS scaling, Cognito callback and "
        "Service Connect fixups), run post_cdk_build_quickstart.py instead."
    )


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(
            REGIONAL_STACK, "CloudFrontDistributionURL", 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)


def maybe_complete_agentcore_cdk_deploy(
    values: Dict[str, str],
    args: argparse.Namespace,
    *,
    assume_yes: bool,
) -> None:
    """Phase 2: create the AgentCore runtime once its ECR image exists, then wire URL.

    Uses the bedrock-agentcore-control API (boto3) rather than a second
    ``cdk deploy``. A second deploy of RedactionStack re-runs the precheck and
    flips the stack's own managed roles/buckets/pool to imports, deleting them
    (removal policy DESTROY on the demo profile) — so the runtime is created
    out-of-band to avoid that footgun.
    """
    if values.get("AGENTCORE_CDK_DEPLOY") != "True":
        return
    if (values.get("AGENTCORE_RUNTIME_URL") or "").strip():
        # Already wired in a previous run; nothing to do.
        return

    print(
        "\n--- AgentCore phase 2 (create runtime from ECR image, no stack update) ---"
    )
    try:
        from cdk_config import (
            CODEBUILD_AGENTCORE_PROJECT_NAME,
            ECR_AGENTCORE_REPO_NAME,
        )
        from cdk_post_deploy import (
            create_agentcore_runtime_from_ecr,
            wait_for_agentcore_ecr_image,
        )
    except ImportError as exc:
        print(f"Skipping AgentCore phase 2: {exc}")
        return

    build_id = (values.get("AGENTCORE_LAST_CODEBUILD_ID") or "").strip() or None
    image_ready = wait_for_agentcore_ecr_image(
        repository_name=ECR_AGENTCORE_REPO_NAME,
        codebuild_project=CODEBUILD_AGENTCORE_PROJECT_NAME,
        build_id=build_id,
    )
    if not image_ready:
        print(
            "\nAgentCore ECR image is not ready yet. Once the image is in ECR "
            "(CodeBuild succeeds or you push it manually), complete phase 2 with:\n"
            "  python cdk_install.py --complete-agentcore"
        )
        return

    if not assume_yes and values.get("USE_ECS_EXPRESS_MODE") != "True":
        proceed = ask_yes_no(
            "Create the Bedrock AgentCore Runtime from the ECR image now?",
            default=True,
        )
        if not proceed:
            print(
                "Skipped AgentCore runtime creation. When the image is in ECR, "
                "complete phase 2 with:\n"
                "  python cdk_install.py --complete-agentcore"
            )
            return

    if args.config_only:
        print(
            "Config-only run: AgentCore runtime not created. Complete phase 2 with:\n"
            "  python cdk_install.py --complete-agentcore"
        )
        return

    create_agentcore_runtime_from_ecr(stack_name=REGIONAL_STACK)


# ---------------------------------------------------------------------------
# Agentic redaction configuration
# ---------------------------------------------------------------------------


def apply_agent_orchestrator_cli_flags(
    args: argparse.Namespace, answers: InstallAnswers
) -> None:
    if getattr(args, "agent_orchestrator", None):
        answers.agent_orchestrator = normalize_agent_orchestrator(
            args.agent_orchestrator
        )
    if getattr(args, "enable_agentcore_runtime", False):
        answers.enable_agentcore_runtime = True
        answers.agent_orchestrator = "agentcore"
    if getattr(args, "enable_agentcore_cdk_deploy", False):
        answers.enable_agentcore_cdk_deploy = True
        answers.enable_agentcore_runtime = True
        answers.agent_orchestrator = "agentcore"
        answers.allow_empty_agentcore_url = True
    if getattr(args, "enable_agentcore_cdk_runtime", False):
        answers.enable_agentcore_cdk_runtime = True
        answers.enable_agentcore_cdk_deploy = True
        answers.enable_agentcore_runtime = True
        answers.agent_orchestrator = "agentcore"
        answers.allow_empty_agentcore_url = True
    if getattr(args, "agentcore_runtime_url", None):
        answers.agentcore_runtime_url = normalize_agentcore_runtime_url(
            str(args.agentcore_runtime_url)
        )
    if getattr(args, "agentcore_api_key", None):
        answers.agentcore_api_key = str(args.agentcore_api_key).strip()
    if getattr(args, "agentcore_model", None):
        answers.agentcore_bedrock_model = normalize_agentcore_bedrock_model(
            str(args.agentcore_model)
        )


def configure_agent_orchestrator_options(
    answers: InstallAnswers,
    args: argparse.Namespace,
    *,
    interactive: bool,
    assume_yes: bool,
) -> None:
    """Prompt for Pi vs LangGraph vs AgentCore orchestration when agentic redaction is enabled."""
    if not answers.agentic_enabled:
        return

    apply_agent_orchestrator_cli_flags(args, answers)
    apply_demo_agentcore_orchestrator_defaults(answers, args)

    if not interactive or assume_yes:
        if answers.enable_agentcore_runtime:
            answers.agent_orchestrator = "agentcore"
        elif (
            answers.profile == "demo"
            and answers.enable_agentic_express
            and not getattr(args, "agent_orchestrator", None)
        ):
            answers.agent_orchestrator = "agentcore"
            answers.enable_agentcore_runtime = True
        if (
            answers.agent_orchestrator == "agentcore"
            and not (answers.agentcore_runtime_url or "").strip()
            and (
                answers.enable_agentcore_cdk_deploy
                or (
                    answers.profile == "demo"
                    and not getattr(args, "agentcore_runtime_url", None)
                )
            )
        ):
            answers.enable_agentcore_cdk_deploy = True
            answers.allow_empty_agentcore_url = True
        if answers.agent_orchestrator == "agentcore":
            finalize_agentcore_bedrock_model(
                answers, args, interactive=False, assume_yes=assume_yes
            )
        return

    if args.agent_orchestrator:
        if answers.agent_orchestrator == "agentcore":
            finalize_agentcore_bedrock_model(
                answers, args, interactive=interactive, assume_yes=assume_yes
            )
        return

    demo_default_idx = 2 if answers.profile == "demo" else 2
    print("\n--- Agent orchestration backend ---")
    idx = ask_choice(
        "Which backend should power the agentic Gradio UI?",
        [
            "Pi coding agent (bash + skills; most open with tools and file access in container)",
            "LangGraph (curated Python tools only; no shell; less agent freedom but more secure)",
            "Bedrock AgentCore (demo default — Gradio proxies to AgentCore runtime URL; most secure)",
        ],
        default_index=demo_default_idx,
    )
    answers.agent_orchestrator = AGENT_ORCHESTRATOR_CHOICES[idx]

    if answers.agent_orchestrator == "agentcore":
        answers.enable_agentcore_runtime = True
        finalize_agentcore_bedrock_model(
            answers, args, interactive=True, assume_yes=assume_yes
        )
        use_cdk_native = False
        if getattr(args, "enable_agentcore_cdk_deploy", False) or getattr(
            args, "enable_agentcore_cdk_runtime", False
        ):
            use_cdk_native = True
        elif not getattr(args, "agentcore_runtime_url", None):
            if answers.profile == "demo":
                use_cdk_native = True
                print("\n--- AgentCore runtime (demo) ---")
                print(AGENTCORE_CDK_DEPLOY_SUMMARY)
            else:
                print("\n--- AgentCore runtime deployment ---")
                use_cdk_native = ask_yes_no(
                    "Deploy the AgentCore runtime via CDK (CodeBuild -> ECR -> Runtime)? "
                    "(recommended; no local agentcore CLI required)",
                    default=False,
                )
        if use_cdk_native:
            answers.enable_agentcore_cdk_deploy = True
            answers.allow_empty_agentcore_url = True
            if getattr(args, "enable_agentcore_cdk_runtime", False):
                answers.enable_agentcore_cdk_runtime = True
        else:
            answers.agentcore_runtime_url = normalize_agentcore_runtime_url(
                ask(
                    "AgentCore runtime URL (from `agentcore status` invocationUrl — "
                    "base URL only; do not include /invocations or a trailing slash)",
                    answers.agentcore_runtime_url,
                )
            )
            if not answers.agentcore_runtime_url:
                print(
                    "\nNo runtime URL yet — CDK can proceed, but the agent UI will not work "
                    "until you deploy the runtime and set AGENTCORE_RUNTIME_URL."
                )
                print(AGENTCORE_MANUAL_DEPLOY_CHECKLIST)
                if ask_yes_no(
                    "Continue without AGENTCORE_RUNTIME_URL for now?",
                    default=answers.profile == "demo",
                ):
                    answers.allow_empty_agentcore_url = True
        print(
            "On AWS ECS the task role uses IAM SigV4 for AgentCore — "
            "AGENTCORE_API_KEY is usually not needed unless the runtime uses "
            "CUSTOM_JWT bearer auth."
        )
        if ask_yes_no(
            "Set AGENTCORE_API_KEY bearer token in config?",
            default=bool(answers.agentcore_api_key),
        ):
            answers.agentcore_api_key = ask(
                "AgentCore API key (paste token; not auto-generated — leave blank to skip)",
                answers.agentcore_api_key,
            ).strip()
    elif answers.agent_orchestrator == "langgraph":
        print(
            "LangGraph: orchestration runs inside the agent container using curated tools "
            "(AGENT_DEFAULT_PROVIDER / Bedrock still applies for the LLM)."
        )


def apply_agentic_cli_flags(args: argparse.Namespace, answers: InstallAnswers) -> None:
    apply_agent_orchestrator_cli_flags(args, answers)
    if getattr(args, "enable_agentic", False):
        preset = merge_preset(answers.profile, answers.custom_overrides)
        if preset.get("USE_ECS_EXPRESS_MODE") == "True":
            answers.enable_agentic_express = True
        else:
            answers.enable_agentic_legacy = True
            answers.enable_service_connect = True
    if getattr(args, "enable_agentic_express", False):
        answers.enable_agentic_express = True
    if getattr(args, "enable_agentic_legacy", False):
        answers.enable_agentic_legacy = True
        answers.enable_service_connect = True
    apply_demo_agentcore_orchestrator_defaults(answers, args)
    if getattr(args, "agentic_alb_routing", None):
        answers.agentic_alb_routing = args.agentic_alb_routing
    if getattr(args, "agentic_path_prefix", None):
        answers.agentic_alb_path_prefix = args.agentic_path_prefix
    if getattr(args, "agentic_host_header", None):
        answers.agentic_alb_host_header = args.agentic_host_header
    if getattr(args, "agentic_listener_priority", None):
        answers.agentic_alb_listener_rule_priority = args.agentic_listener_priority
    if getattr(args, "agentic_gradio_port", None):
        answers.agentic_gradio_port = args.agentic_gradio_port
    if getattr(args, "sc_discovery_name", None):
        answers.sc_discovery_name = args.sc_discovery_name
    if getattr(args, "pi_provider", None):
        answers.pi_default_provider = args.pi_provider
    if getattr(args, "skip_pi_agent_env", False):
        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_agentic_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_agentic_cli_flags(args, answers)

    if not answers.agentic_enabled and interactive:
        if use_express:
            label = "Deploy agentic redaction (second Gradio app on Express, dedicated HTTPS URL)?"
            answers.enable_agentic_express = ask_yes_no(label, default=False)
        else:
            label = "Deploy agentic redaction (second Gradio app on legacy Fargate + Service Connect)?"
            if ask_yes_no(label, default=False):
                answers.enable_agentic_legacy = True
                answers.enable_service_connect = True

    if not answers.agentic_enabled:
        return

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

    if (
        not answers.agentic_alb_listener_rule_priority
        and not answers.enable_agentic_express
    ):
        answers.agentic_alb_listener_rule_priority = default_agentic_listener_priority(
            use_cloudfront
        )

    if answers.enable_agentic_express:
        if interactive and not assume_yes:
            print("\n--- Agentic redaction (ECS Express) ---")
            answers.sc_discovery_name = ask(
                "Service Connect discovery name for main app",
                answers.sc_discovery_name,
            )
            answers.agentic_gradio_port = ask(
                "Agent Gradio port", answers.agentic_gradio_port
            )
            if ask_yes_no(
                f"Write/update {PI_AGENT_ENV_PATH.name} for AWS ECS (DOC_REDACTION_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(
                        "agent.env exists — replace file from example template?",
                        default=False,
                    )
            else:
                answers.write_pi_agent_env = False
        print(
            f"Agentic redaction: Express (dedicated HTTPS endpoint per service); "
            f"Service Connect discovery={answers.sc_discovery_name}; "
            f"orchestrator={normalize_agent_orchestrator(answers.agent_orchestrator)}"
        )
        return

    if interactive and not (
        args.agentic_alb_routing
        or args.agentic_path_prefix
        or args.agentic_host_header
        or assume_yes
    ):
        print("\n--- Agentic redaction 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 (AGENT_ALB_HOST_HEADER)",
                "Both path prefix and hostname",
            ],
            default_index=0,
        )
        answers.agentic_alb_routing = AGENTIC_ALB_ROUTING_MODES[ridx]

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

        if use_cloudfront:
            default_pri = default_agentic_listener_priority(True)
            answers.agentic_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.agentic_gradio_port = ask(
            "Agent Gradio port", answers.agentic_gradio_port
        )

        if ask_yes_no(
            f"Write/update {PI_AGENT_ENV_PATH.name} for AWS ECS (DOC_REDACTION_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(
                    "agent.env exists — replace file from example template?",
                    default=False,
                )
        else:
            answers.write_pi_agent_env = False

    print(
        f"Agentic redaction: "
        f"{'Express' if answers.enable_agentic_express else 'legacy Fargate'}, "
        f"routing={answers.agentic_alb_routing}, "
        f"prefix={normalize_agentic_path_prefix(answers.agentic_alb_path_prefix)}, "
        f"orchestrator={normalize_agent_orchestrator(answers.agent_orchestrator)}"
    )


# ---------------------------------------------------------------------------
# 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"] = "False"
        answers.custom_overrides["ENABLE_RESOURCE_DELETE_PROTECTION"] = (
            "True" if ask_yes_no("Enable delete protection?", True) else "False"
        )
        answers.custom_overrides["ENABLE_APPREGISTRY"] = (
            "True"
            if ask_yes_no(
                "Enable AppRegistry (AWS Console myApplications)? AWS is retiring "
                "creation of new AppRegistry Applications, so this may fail on new "
                "accounts",
                default=False,
            )
            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,
            )

    # Delete/termination protection. The production preset defaults this ON, but
    # let the user opt out interactively (custom already prompts above). This
    # single flag drives stack termination protection and every resource-level
    # deletion protection / RETAIN removal policy via
    # ENABLE_RESOURCE_DELETE_PROTECTION.
    if interactive and answers.profile == "production":
        answers.custom_overrides["ENABLE_RESOURCE_DELETE_PROTECTION"] = (
            "True"
            if ask_yes_no(
                "Enable delete/termination protection for the stack and its "
                "resources (recommended for production)?",
                default=True,
            )
            else "False"
        )

    # A --delete-protection flag overrides the profile default (and any prompt),
    # so non-interactive runs can select the behaviour explicitly.
    delete_protection_cli = getattr(args, "delete_protection", None)
    if delete_protection_cli in ("on", "off"):
        answers.custom_overrides["ENABLE_RESOURCE_DELETE_PROTECTION"] = (
            "True" if delete_protection_cli == "on" else "False"
        )

    # AppRegistry (AWS Console myApplications). Default OFF because AWS is retiring
    # creation of new AppRegistry Applications; offer to opt in interactively for the
    # demo/production profiles (custom already prompts above).
    if interactive and answers.profile in ("demo", "production"):
        answers.custom_overrides["ENABLE_APPREGISTRY"] = (
            "True"
            if ask_yes_no(
                "Enable AppRegistry (AWS Console myApplications)? AWS is retiring "
                "creation of new AppRegistry Applications, so leave this off unless "
                "your account can still create them",
                default=False,
            )
            else "False"
        )

    # A --appregistry flag overrides the profile default (and any prompt), so
    # non-interactive runs can select the behaviour explicitly.
    appregistry_cli = getattr(args, "appregistry", None)
    if appregistry_cli in ("on", "off"):
        answers.custom_overrides["ENABLE_APPREGISTRY"] = (
            "True" if appregistry_cli == "on" else "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()}-Redaction-"
    answers.cdk_prefix = args.cdk_prefix or (
        ask("CDK resource prefix (e.g. MyOrg-Redaction-)", 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):
        configure_agentic_options(
            answers,
            args,
            interactive=interactive,
            assume_yes=assume_yes,
        )
    else:
        answers.enable_s3_batch = True

    # Advanced add-ons (without agentic redaction)
    is_express = use_express and not answers_use_headless(answers)
    if answers_use_headless(answers) and interactive:
        cpu = ask("ECS task CPU (units; 1024 = 1 vCPU)", answers.ecs_cpu)
        if cpu:
            answers.ecs_cpu = cpu
        mem = ask("ECS task memory (MB)", answers.ecs_memory)
        if mem:
            answers.ecs_memory = mem
    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 agentic redaction)?", False
                )
            answers.enable_s3_batch = ask_yes_no("Enable S3 batch ECS trigger?", False)
        cpu = ask("ECS task CPU (units; 1024 = 1 vCPU)", answers.ecs_cpu)
        if cpu:
            answers.ecs_cpu = cpu
        mem = ask("ECS task memory (MB)", answers.ecs_memory)
        if mem:
            answers.ecs_memory = mem

    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 doc_redaction",
        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(
        "--complete-agentcore",
        action="store_true",
        help="AgentCore two-phase deploy, phase 2 only: after the AgentCore runtime "
        "image is in ECR, create the Bedrock runtime via the bedrock-agentcore "
        "API (boto3) and sync the runtime URL. No cdk deploy / no stack update. "
        "Uses existing cdk_config.env; skips the wizard and existing-stack guard.",
    )
    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="Explicit opt-in: if doc_redaction stacks already exist, delete them "
        "before a clean initial deploy (disables termination protection). Without "
        "this flag cdk_install aborts when a stack exists (it never updates in place).",
    )
    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(
        "--s3-malware-scan-bucket",
        help="S3 malware scan staging bucket name (globally unique)",
    )
    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("--cert-arn", help="ACM certificate ARN (production)")
    p.add_argument("--domain", help="SSL certificate domain (production)")
    p.add_argument(
        "--delete-protection",
        choices=("on", "off"),
        help="Override stack termination + resource deletion protection "
        "(default: on for production, off for demo/headless). Applies to all "
        "resources governed by ENABLE_RESOURCE_DELETE_PROTECTION.",
    )
    p.add_argument(
        "--appregistry",
        choices=("on", "off"),
        help="Override AWS Console myApplications (AppRegistry) creation "
        "(default: off). AWS is retiring creation of new AppRegistry "
        "Applications; sets ENABLE_APPREGISTRY.",
    )
    agentic = p.add_argument_group("Agentic redaction (second Gradio app)")
    agentic.add_argument(
        "--enable-agentic",
        "--enable-pi",
        dest="enable_agentic",
        action="store_true",
        help="Enable agentic redaction for current profile (Express on demo, legacy on production)",
    )
    agentic.add_argument(
        "--enable-agentic-express",
        "--enable-pi-express",
        dest="enable_agentic_express",
        action="store_true",
        help="Enable agentic redaction on ECS Express Mode (demo)",
    )
    agentic.add_argument(
        "--enable-agentic-legacy",
        "--enable-pi-legacy",
        dest="enable_agentic_legacy",
        action="store_true",
        help="Enable agentic redaction on legacy Fargate (implies Service Connect)",
    )
    agentic.add_argument(
        "--agentic-alb-routing",
        "--pi-alb-routing",
        dest="agentic_alb_routing",
        choices=AGENTIC_ALB_ROUTING_MODES,
        help="ALB routing mode for agentic UI (default: path)",
    )
    agentic.add_argument(
        "--agentic-path-prefix",
        "--pi-path-prefix",
        dest="agentic_path_prefix",
        default="",
        help="Agentic UI path prefix (default: /agent)",
    )
    agentic.add_argument(
        "--agentic-host-header",
        "--pi-host-header",
        dest="agentic_host_header",
        default="",
        help="Dedicated hostname for agentic mode when routing is host or both",
    )
    agentic.add_argument(
        "--agentic-listener-priority",
        "--pi-listener-priority",
        dest="agentic_listener_priority",
        default="",
        help="ALB listener rule priority for agentic mode (default 3; priorities 1–2 reserved)",
    )
    agentic.add_argument(
        "--agentic-gradio-port",
        "--pi-gradio-port",
        dest="agentic_gradio_port",
        default="",
        help="Agentic Gradio listen port (default 7862)",
    )
    agentic.add_argument(
        "--sc-discovery-name",
        default="",
        help="Service Connect name for main app (default redaction)",
    )
    agentic.add_argument(
        "--pi-provider",
        default="",
        help="AGENT_DEFAULT_PROVIDER for agent.env (default amazon-bedrock)",
    )
    agentic.add_argument(
        "--skip-pi-agent-env",
        action="store_true",
        help="Do not write config/agent.env",
    )
    agentic.add_argument(
        "--agent-orchestrator",
        choices=AGENT_ORCHESTRATOR_CHOICES,
        help="Agent orchestration backend (demo+--enable-agentic default: agentcore)",
    )
    agentic.add_argument(
        "--enable-agentcore-cdk-deploy",
        action="store_true",
        help=(
            "CDK-native AgentCore phase 1 only (ARM64 CodeBuild + ECR). "
            "Demo + --enable-agentic sets this automatically; phase 2 runs after "
            "the image build without this flag."
        ),
    )
    agentic.add_argument(
        "--enable-agentcore-cdk-runtime",
        action="store_true",
        help=(
            "Force AgentCore phase 2 immediately (Bedrock runtime from ECR). "
            "Usually unnecessary — re-run cdk_install.py after the image build instead."
        ),
    )
    agentic.add_argument(
        "--enable-agentcore-runtime",
        action="store_true",
        help="Use Bedrock AgentCore (sets AGENT_ORCHESTRATOR=agentcore)",
    )
    agentic.add_argument(
        "--agentcore-runtime-url",
        default="",
        help=(
            "AgentCore runtime base URL from `agentcore status` invocationUrl "
            "(no /invocations suffix, no trailing slash)"
        ),
    )
    agentic.add_argument(
        "--agentcore-api-key",
        default="",
        help=(
            "Optional bearer token for CUSTOM_JWT AgentCore auth (not auto-generated; "
            "usually omitted on AWS ECS where IAM SigV4 is used)"
        ),
    )
    agentic.add_argument(
        "--agentcore-model",
        default="",
        help=(
            "Bedrock model id for the AgentCore runtime engine (fixed at runtime "
            f"creation). Default: {AGENTCORE_DEFAULT_BEDROCK_MODEL}. Common choices: "
            + ", ".join(m for m, _ in AGENTCORE_BEDROCK_MODEL_CHOICES)
        ),
    )
    agentic.add_argument(
        "--skip-bedrock-model-check",
        action="store_true",
        help=(
            "Skip validating that the calling AWS identity has access to the chosen "
            "AgentCore Bedrock model."
        ),
    )
    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 _run_complete_agentcore(args: argparse.Namespace) -> int:
    """Phase 2 of the AgentCore two-phase deploy — WITHOUT a stack update.

    Assumes the AgentCore runtime image is already in ECR (built by CodeBuild or
    pushed manually) and that phase 1 created the execution role. Creates the
    Bedrock AgentCore runtime via the bedrock-agentcore-control API (boto3), then
    derives ``AGENTCORE_RUNTIME_URL``, patches config, re-uploads ``agent.env``
    and recycles the agent Express service.

    This deliberately does NOT run ``cdk deploy``. A second ``cdk deploy`` of
    RedactionStack would re-run the precheck and flip the stack's own managed
    roles/buckets/pool to imports, deleting them (removal policy DESTROY on the
    demo profile). Creating the runtime out-of-band avoids that footgun.
    """
    if not ENV_PATH.is_file():
        raise SystemExit(
            f"No config at {ENV_PATH}. Run the initial deploy first, then re-run "
            "with --complete-agentcore once the AgentCore image is in ECR."
        )
    values = read_env_file(ENV_PATH)

    if values.get("AGENTCORE_CDK_DEPLOY") != "True":
        raise SystemExit(
            "AGENTCORE_CDK_DEPLOY is not True in cdk_config.env; there is no "
            "AgentCore CDK phase 1 to complete. (Manual/CLI AgentCore deploys do "
            "not use --complete-agentcore — see agent-redact/agentcore/README.md.)"
        )

    from cdk_post_deploy import create_agentcore_runtime_from_ecr

    url = create_agentcore_runtime_from_ecr(stack_name=REGIONAL_STACK)
    return 0 if url else 1


def main(argv: Optional[Sequence[str]] = None) -> int:
    args = build_arg_parser().parse_args(argv)
    CONFIG_DIR.mkdir(parents=True, exist_ok=True)

    print("doc_redaction 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.")

    # AgentCore two-phase deploy, phase 2 only. This is the supported way to
    # complete the runtime after its ECR image exists — it performs a targeted
    # 'cdk deploy RedactionStack' (the one place an existing stack is updated on
    # purpose) without the wizard or the existing-stack guard.
    if args.complete_agentcore:
        return _run_complete_agentcore(args)

    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,
                allow_empty_agentcore_url=answers.allow_empty_agentcore_url,
            )
        )
        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,
            )
        if answers.agentic_enabled:
            write_pi_agent_env_file(
                answers,
                overwrite=answers.overwrite_pi_agent_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)
    cloudfront_enabled = values.get("USE_CLOUDFRONT") == "True"
    cf_domain_before = (values.get("CLOUDFRONT_DOMAIN") or "").strip().lower()
    apply_post_deploy_fixup(values, assume_yes=args.yes)

    # The initial deploy bakes the placeholder domain into the CloudFront
    # response-headers policy (CSP/CORS) because the real *.cloudfront.net domain
    # is only known after the distribution exists (a self-reference would be a
    # circular dependency). The boto3 fixup above already wrote the real domain
    # into the env file and updated the Cognito callback URLs (no stack change).
    #
    # cdk_install intentionally does NOT re-deploy the stack a second time to
    # push the resolved domain into the response-headers policy: in-place stack
    # updates have repeatedly caused imported-resource drift, so this installer
    # performs a single initial deploy only. If the CSP/response-headers policy
    # must reflect the real domain, delete and redeploy the stack fresh (with
    # CLOUDFRONT_DOMAIN already set) rather than updating in place.
    values = read_env_file(ENV_PATH)
    cf_domain_after = (values.get("CLOUDFRONT_DOMAIN") or "").strip().lower()
    cloudfront_domain_resolved = (
        cloudfront_enabled
        and bool(cf_domain_after)
        and "placeholder" not in cf_domain_after
        and cf_domain_after != cf_domain_before
    )
    if cloudfront_domain_resolved:
        print(
            f"CloudFront domain resolved to {cf_domain_after}. Cognito callback "
            "URLs were updated via the post-deploy fixup. The stack's "
            "response-headers policy (CSP/CORS) keeps the value baked in at the "
            "initial deploy — cdk_install does not update the stack in place. "
            "Redeploy the stack fresh if the CSP/CORS domain must change."
        )

    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)
            maybe_complete_agentcore_cdk_deploy(values, args, assume_yes=args.yes)

    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

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

    print("\nDone. Next steps:")
    if (
        values.get("ENABLE_PI_AGENT_EXPRESS_SERVICE") == "True"
        or values.get("ENABLE_PI_AGENT_ECS_SERVICE") == "True"
    ):
        if values.get("ENABLE_PI_AGENT_EXPRESS_SERVICE") == "True":
            print(
                "  - Agent UI: AgenticExpressEndpoint stack output (dedicated Express HTTPS URL)"
            )
        else:
            routing = values.get("AGENT_ALB_ROUTING", "path")
            if routing in ("path", "both"):
                prefix = values.get("AGENT_ALB_PATH_PREFIX", "/agent")
                print(f"  - Agent UI path: {prefix}/ on your app URL")
            if routing in ("host", "both") and values.get("AGENT_ALB_HOST_HEADER"):
                print(
                    f"  - Agent DNS: CNAME {values['AGENT_ALB_HOST_HEADER']} -> "
                    f"{values.get('CLOUDFRONT_DOMAIN') or 'ALB/Express endpoint'}"
                )
        if PI_AGENT_ENV_PATH.is_file():
            print(
                f"  - Agent runtime config: {PI_AGENT_ENV_PATH} (uploaded by quickstart)"
            )
    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}")
        if values.get("CLOUDFRONT_AUTH_MODE") == "magic-link":
            print(
                "  - After deploy: open RedactionLoginUrl from stack outputs "
                "(?key= unlock), then browse via RedactionUrl / CloudFrontDistributionURL"
            )
    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())