#!/usr/bin/env python3 """ Interactive CDK installer for llm_topic_modeller. Walks through demo vs production deployment profiles, writes config/cdk_config.env and cdk.json, optionally runs cdk deploy, post-deploy Cognito callback fixups (API, no second deploy), and post_cdk_build_quickstart.py. Usage examples:: # Full interactive install + deploy + quickstart python cdk_install.py # Demo sandbox, config only python cdk_install.py --profile demo --config-only --yes # Production with flags (partial non-interactive) python cdk_install.py --profile production --vpc-name my-vpc \\ --cert-arn arn:aws:acm:eu-west-2:123:certificate/abc \\ --domain llm-topic.example.com --yes # Redeploy using existing config, skip quickstart python cdk_install.py --deploy-only --skip-quickstart # Remove existing stacks before a clean install (non-interactive) python cdk_install.py --profile headless --vpc-name my-vpc \\ --force-delete-stacks --yes # Headless batch (S3 → Lambda → one-shot ECS direct mode) python cdk_install.py --profile headless --vpc-name my-vpc --yes python cdk_install.py --profile production --headless --vpc-name my-vpc --yes # Optional scheduled DynamoDB usage log export to S3 python cdk_install.py --profile demo --vpc-name my-vpc --yes \\ --dynamo-usage-log-export --dynamo-export-schedule-time 06:00 \\ --dynamo-export-schedule-days weekdays # Headless with S3 output email alerts + IAM reader user python cdk_install.py --profile headless --vpc-name my-vpc --yes \\ --headless-output-notifications \\ --headless-notify-email analyst@example.com # Enable Bedrock model invocation logging (S3 log bucket + CloudWatch) python cdk_install.py --profile headless --vpc-name my-vpc --yes \\ --bedrock-model-invocation-logging # Prefer Resource Groups + tags; leave AppRegistry off (default) python cdk_install.py --profile production --vpc-name my-vpc --yes \\ --application-resource-group --application-name my-prefix-llm-topic-modeller # Opt in to legacy AppRegistry / myApplications (existing customers only) python cdk_install.py --profile demo --vpc-name my-vpc --yes --appregistry """ from __future__ import annotations import argparse import ipaddress import json import os import re import secrets import shutil import subprocess import sys from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional, Sequence, Tuple CDK_DIR = Path(__file__).resolve().parent REPO_ROOT = CDK_DIR.parent CONFIG_DIR = CDK_DIR / "config" ENV_PATH = CONFIG_DIR / "cdk_config.env" APP_CONFIG_ENV_PATH = CONFIG_DIR / "app_config.env" APP_CONFIG_ENV_EXAMPLE = CONFIG_DIR / "app_config.env.example" PI_AGENT_ENV_PATH = CONFIG_DIR / "pi_agent.env" PI_AGENT_ENV_EXAMPLE = REPO_ROOT / "config" / "pi_agent.env.example" PI_ALB_ROUTING_MODES = ("path", "host", "both") CDK_JSON_PATH = CDK_DIR / "cdk.json" CDK_JSON_EXAMPLE = CDK_DIR / "cdk.json.example" QUICKSTART_SCRIPT = CDK_DIR / "post_cdk_build_quickstart.py" REGIONAL_STACK = "SummarisationStack" CLOUDFRONT_STACK = "SummarisationStackCloudfront" CLOUDFRONT_STACK_REGION = "us-east-1" APPREGISTRY_STACK_SUFFIX = "AppRegistryStack" DEMO_PRESET: Dict[str, str] = { "USE_ECS_EXPRESS_MODE": "True", "ECS_EXPRESS_USE_PUBLIC_SUBNETS": "True", "ENABLE_ECS_VPC_INTERFACE_ENDPOINTS": "True", "USE_CLOUDFRONT": "False", "RUN_USEAST_STACK": "False", "ENABLE_RESOURCE_DELETE_PROTECTION": "False", "ENABLE_APPREGISTRY": "False", "ENABLE_APPLICATION_RESOURCE_GROUP": "True", "ACM_SSL_CERTIFICATE_ARN": "", "SSL_CERTIFICATE_DOMAIN": "", } PRODUCTION_PRESET: Dict[str, str] = { "USE_ECS_EXPRESS_MODE": "False", "USE_CLOUDFRONT": "True", "RUN_USEAST_STACK": "True", "ENABLE_RESOURCE_DELETE_PROTECTION": "True", "ENABLE_APPREGISTRY": "False", "ENABLE_APPLICATION_RESOURCE_GROUP": "True", } HEADLESS_PRESET: Dict[str, str] = { "USE_ECS_EXPRESS_MODE": "False", "ECS_EXPRESS_USE_PUBLIC_SUBNETS": "True", "USE_CLOUDFRONT": "False", "RUN_USEAST_STACK": "False", "ENABLE_RESOURCE_DELETE_PROTECTION": "False", "ENABLE_APPREGISTRY": "False", "ENABLE_APPLICATION_RESOURCE_GROUP": "True", "ENABLE_HEADLESS_DEPLOYMENT": "True", "ENABLE_S3_BATCH_ECS_TRIGGER": "True", "COGNITO_AUTH": "False", "ACM_SSL_CERTIFICATE_ARN": "", "SSL_CERTIFICATE_DOMAIN": "", } DEFAULT_CDK_JSON_CONTEXT: Dict[str, Any] = { "@aws-cdk/aws-ec2:restrictDefaultSecurityGroup": False, } # --------------------------------------------------------------------------- # Wizard helpers # --------------------------------------------------------------------------- def ask(prompt: str, default: str = "") -> str: suffix = f" [{default}]" if default else "" value = input(f"{prompt}{suffix}: ").strip() return value or default def ask_yes_no(prompt: str, default: bool = True) -> bool: default_label = "Y/n" if default else "y/N" while True: raw = input(f"{prompt} ({default_label}): ").strip().lower() if not raw: return default if raw in ("y", "yes"): return True if raw in ("n", "no"): return False print("Please enter y or n.") def ask_choice(prompt: str, options: Sequence[str], default_index: int = 0) -> int: print(prompt) for i, opt in enumerate(options): marker = "*" if i == default_index else " " print(f" {marker} {i + 1}) {opt}") while True: raw = input( f"Choice [1-{len(options)}] (default {default_index + 1}): " ).strip() if not raw: return default_index try: idx = int(raw) - 1 if 0 <= idx < len(options): return idx except ValueError: pass print("Invalid choice.") def timestamp() -> str: return datetime.now().strftime("%Y%m%d-%H%M%S") def backup_file(path: Path) -> Optional[Path]: if not path.is_file(): return None backup = path.with_suffix(path.suffix + f".bak.{timestamp()}") shutil.copy2(path, backup) print(f"Backed up {path.name} -> {backup.name}") return backup # --------------------------------------------------------------------------- # Python / cdk.json # --------------------------------------------------------------------------- def _venv_python_paths() -> List[Path]: candidates: List[Path] = [] virtual_env = os.environ.get("VIRTUAL_ENV", "").strip() if virtual_env: root = Path(virtual_env) if os.name == "nt": candidates.append(root / "Scripts" / "python.exe") else: candidates.append(root / "bin" / "python") rel_venvs = [ CDK_DIR / ".venv", REPO_ROOT / ".venv", REPO_ROOT.parent / ".venv", ] for venv_root in rel_venvs: if os.name == "nt": candidates.append(venv_root / "Scripts" / "python.exe") else: candidates.append(venv_root / "bin" / "python") for name in ("python3", "python"): found = shutil.which(name) if found: candidates.append(Path(found)) seen: set[str] = set() unique: List[Path] = [] for p in [Path(sys.executable)] + candidates: key = str(p.resolve()) if p.exists() else str(p) if key not in seen: seen.add(key) unique.append(p) return unique def _resolve_node_executable() -> Optional[str]: return shutil.which("node") def _jsii_import_failure_hint(stderr: str) -> str: if "JSONDecodeError" not in stderr and "Expecting value" not in stderr: return "" return ( "\nThis usually means the JSII Node.js helper process failed to start.\n" "On the deployment machine, check:\n" " 1. Node.js is installed and on PATH: node --version (LTS 18–22 recommended)\n" " 2. Reinstall CDK Python deps: pip install --force-reinstall -r requirements.txt\n" " 3. Test manually: python -c \"from aws_cdk import App; print('ok')\"\n" " 4. If still failing, run: set JSII_DEBUG=1 (then retry synth for details)" ) def verify_node_for_jsii() -> str: """Return the node executable path, or exit with guidance if missing.""" node = _resolve_node_executable() if not node: raise SystemExit( "Node.js not found on PATH. aws-cdk-lib (JSII) requires Node.js.\n" "Install Node.js LTS from https://nodejs.org/ and ensure 'node' is on PATH." ) try: result = subprocess.run( [node, "--version"], capture_output=True, text=True, timeout=10, check=False, ) except (OSError, subprocess.TimeoutExpired) as exc: raise SystemExit(f"Could not run Node.js ({node}): {exc}") from exc if result.returncode != 0: raise SystemExit( f"Node.js ({node}) returned exit code {result.returncode}.\n" f"{(result.stderr or result.stdout or '').strip()}" ) version = (result.stdout or result.stderr or "").strip() print(f"Node.js for JSII: {node} ({version})") return node def _python_has_aws_cdk(python_exe: Path) -> Tuple[bool, str]: if not python_exe.is_file(): return False, "not found" cmd = [ str(python_exe), "-c", "import aws_cdk; print(getattr(aws_cdk, '__version__', 'unknown'))", ] try: result = subprocess.run( cmd, cwd=str(CDK_DIR), capture_output=True, text=True, timeout=30, check=False, ) if result.returncode == 0: version = (result.stdout or "").strip().splitlines()[-1] return True, version detail = (result.stderr or result.stdout or "import failed").strip() hint = _jsii_import_failure_hint(detail) if hint: detail = f"{detail[:400]}{hint}" return False, detail[:800] except (OSError, subprocess.TimeoutExpired) as exc: return False, str(exc) def discover_python_candidates() -> List[Tuple[Path, str]]: found: List[Tuple[Path, str]] = [] for candidate in _venv_python_paths(): ok, detail = _python_has_aws_cdk(candidate) if ok: found.append((candidate.resolve(), detail)) return found def resolve_python_executable( override: Optional[str] = None, interactive: bool = True, assume_yes: bool = False, ) -> Path: if override: path = Path(override).resolve() ok, detail = _python_has_aws_cdk(path) if not ok: raise SystemExit( f"Python at {path} cannot import aws-cdk-lib: {detail}\n" "Install CDK deps: pip install -r requirements.txt" ) print(f"Using Python: {path} (aws-cdk-lib {detail})") return path candidates = discover_python_candidates() if not candidates: raise SystemExit( "No Python interpreter with aws-cdk-lib found.\n" "Activate your venv and run: pip install -r requirements.txt\n" "Or pass --python /path/to/python" ) if len(candidates) == 1 or assume_yes or not interactive: path, version = candidates[0] print(f"Using Python: {path} (aws-cdk-lib {version})") return path print("Python interpreters with aws-cdk-lib:") labels = [f"{p} (aws-cdk-lib {v})" for p, v in candidates] idx = ask_choice("Select Python for cdk.json", labels, default_index=0) path, version = candidates[idx] print(f"Selected: {path} (aws-cdk-lib {version})") return path def format_cdk_app_command(python_exe: Path) -> str: """Format the cdk.json ``app`` command for the current OS.""" py = str(python_exe) if os.name == "nt": py = py.replace("/", "\\") return f"{py} app.py" def load_cdk_json_context(existing_path: Path) -> Dict[str, Any]: if existing_path.is_file(): try: data = json.loads(existing_path.read_text(encoding="utf-8")) ctx = data.get("context") if isinstance(ctx, dict): return dict(ctx) except json.JSONDecodeError: pass if CDK_JSON_EXAMPLE.is_file(): try: data = json.loads(CDK_JSON_EXAMPLE.read_text(encoding="utf-8")) ctx = data.get("context") if isinstance(ctx, dict): return dict(ctx) except json.JSONDecodeError: pass return dict(DEFAULT_CDK_JSON_CONTEXT) def write_cdk_json( python_exe: Path, *, force: bool = False, skip: bool = False, ) -> Optional[Path]: if skip: if not CDK_JSON_PATH.is_file(): raise SystemExit( f"{CDK_JSON_PATH.name} not found; cannot use --skip-cdk-json." ) print(f"Keeping existing {CDK_JSON_PATH.name} (--skip-cdk-json)") return CDK_JSON_PATH if CDK_JSON_PATH.is_file() and force: backup_file(CDK_JSON_PATH) elif not CDK_JSON_PATH.is_file(): pass else: # Update app only — backup first backup_file(CDK_JSON_PATH) context = load_cdk_json_context(CDK_JSON_PATH) payload = { "app": format_cdk_app_command(python_exe), "output": "cdk.out", "context": context, } CDK_JSON_PATH.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") print(f"Wrote {CDK_JSON_PATH}") return CDK_JSON_PATH def build_cdk_subprocess_env( values: Optional[Dict[str, str]] = None, ) -> Dict[str, str]: """ Environment for child Python / CDK processes. Merges config/cdk_config.env over the current process environment so stale empty defaults from an earlier cdk_config import during the wizard do not block load_dotenv (override=False) in app.py. """ env = os.environ.copy() env["CDK_CONFIG_PATH"] = str(ENV_PATH) cdk_folder = "" if values: cdk_folder = (values.get("CDK_FOLDER") or "").strip() if not cdk_folder and ENV_PATH.is_file(): cdk_folder = (read_env_file(ENV_PATH).get("CDK_FOLDER") or "").strip() if not cdk_folder: cdk_folder = str(CDK_DIR).replace("\\", "/") + "/" env["CDK_FOLDER"] = cdk_folder file_values: Dict[str, str] = {} if ENV_PATH.is_file(): file_values = read_env_file(ENV_PATH) elif values: file_values = dict(values) for key, val in file_values.items(): env[key] = str(val) return env def apply_cdk_runtime_env(values: Dict[str, str]) -> None: """Expose written config to app.py imports (smoke test, cdk synth/deploy).""" os.environ.update(build_cdk_subprocess_env(values)) def smoke_test_python_app(python_exe: Path) -> None: verify_node_for_jsii() cmd = [str(python_exe), "-c", "import aws_cdk; import app"] print("Smoke test: import aws_cdk and app ...") result = subprocess.run( cmd, cwd=str(CDK_DIR), env=build_cdk_subprocess_env(), capture_output=True, text=True, check=False, ) if result.returncode != 0: output = (result.stderr or result.stdout or "").strip() print(output) hint = _jsii_import_failure_hint(output) if hint: print(hint) raise SystemExit( "Python smoke test failed. Fix CDK dependencies on this machine before deploying." ) def run_smoke_test_if_needed( python_exe: Optional[Path], args: argparse.Namespace, ) -> None: if args.config_only or args.skip_cdk_json or not python_exe: return smoke_test_python_app(python_exe) # --------------------------------------------------------------------------- # AWS discovery # --------------------------------------------------------------------------- def get_aws_identity(region: Optional[str] = None) -> Tuple[str, str]: import boto3 session = boto3.Session(region_name=region) sts = session.client("sts") ident = sts.get_caller_identity() account = ident["Account"] resolved_region = ( region or session.region_name or os.environ.get("AWS_REGION") or os.environ.get("AWS_DEFAULT_REGION") or "eu-west-2" ) return account, resolved_region def resolve_cdk_executable() -> str: """Return the AWS CDK CLI on PATH (cdk.cmd on Windows).""" for name in ("cdk", "cdk.cmd", "cdk.exe"): found = shutil.which(name) if found: return found raise SystemExit( "AWS CDK CLI not found on PATH.\n" "Install: npm install -g aws-cdk (or use npx aws-cdk)" ) def check_cdk_cli() -> str: cdk_exe = resolve_cdk_executable() result = subprocess.run( [cdk_exe, "--version"], capture_output=True, text=True, check=False, ) if result.returncode != 0: raise SystemExit( "AWS CDK CLI not found on PATH.\n" "Install: npm install -g aws-cdk (or use npx aws-cdk)" ) version = (result.stdout or result.stderr or "").strip() if version.startswith("Python "): raise SystemExit( "CDK CLI check failed: 'cdk' resolved to the Python interpreter instead " "of the AWS CDK CLI. Install aws-cdk globally (npm install -g aws-cdk) " "and ensure it is on PATH before python.exe." ) return version def cdk_bootstrap_needed(account: str, region: str) -> bool: import boto3 from botocore.exceptions import ClientError cfn = boto3.client("cloudformation", region_name=region) try: cfn.describe_stacks(StackName="CDKToolkit") return False except ClientError as exc: if "does not exist" in str(exc): return True raise def list_vpcs(region: str) -> List[Dict[str, str]]: import boto3 ec2 = boto3.client("ec2", region_name=region) response = ec2.describe_vpcs() vpcs: List[Dict[str, str]] = [] for vpc in response.get("Vpcs", []): name = "" for tag in vpc.get("Tags", []): if tag.get("Key") == "Name": name = tag.get("Value", "") break cidrs = vpc_cidr_blocks_from_describe(vpc) vpcs.append( { "id": vpc["VpcId"], "name": name or vpc["VpcId"], "cidr": cidrs[0] if cidrs else "", "cidrs": ",".join(cidrs), } ) return sorted(vpcs, key=lambda x: x["name"]) def vpc_cidr_blocks_from_describe(vpc: Dict[str, Any]) -> List[str]: """Primary and associated IPv4 CIDR blocks for one ``describe_vpcs`` entry.""" blocks: List[str] = [] seen: set[str] = set() primary = (vpc.get("CidrBlock") or "").strip() if primary and primary not in seen: seen.add(primary) blocks.append(primary) for assoc in vpc.get("CidrBlockAssociationSet") or []: cidr = (assoc.get("CidrBlock") or "").strip() if cidr and cidr not in seen: seen.add(cidr) blocks.append(cidr) return blocks def list_vpc_cidr_blocks_in_region(region: str) -> List[str]: """All IPv4 VPC CIDR blocks in the account/region (primary + associations).""" import boto3 ec2 = boto3.client("ec2", region_name=region) blocks: List[str] = [] seen: set[str] = set() for vpc in ec2.describe_vpcs().get("Vpcs", []): for cidr in vpc_cidr_blocks_from_describe(vpc): if cidr not in seen: seen.add(cidr) blocks.append(cidr) return blocks VPC_CIDR_PREFIX_LEN = 24 SUBNET_CIDR_PREFIX_LEN = 28 # ECS Express provisions a managed ALB in each public subnet (8+ free IPs required). # /28 subnets (~11 usable IPs) exhaust quickly with VPC interface endpoints and tasks. EXPRESS_PUBLIC_SUBNET_CIDR_PREFIX_LEN = 27 PUBLIC_SUBNET_CIDR_PREFIX_LEN = 26 PRIVATE_SUBNET_CIDR_PREFIX_LEN = 28 EXPRESS_ALB_MIN_SUBNET_PREFIX_LEN = 27 _VPC_CIDR_SEARCH_SUPERNETS = ("10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16") def parse_ipv4_networks(cidrs: Sequence[str]) -> List[ipaddress.IPv4Network]: networks: List[ipaddress.IPv4Network] = [] for cidr in cidrs: raw = (cidr or "").strip() if not raw: continue try: networks.append(ipaddress.ip_network(raw, strict=False)) except ValueError: continue return networks def suggest_vpc_cidr_block( existing_vpc_cidrs: Sequence[str], *, prefix_len: int = VPC_CIDR_PREFIX_LEN, search_supernets: Sequence[str] = _VPC_CIDR_SEARCH_SUPERNETS, ) -> str: """Return the lowest non-overlapping ``prefix_len`` block in RFC1918 space.""" occupied = parse_ipv4_networks(existing_vpc_cidrs) for supernet_cidr in search_supernets: supernet = ipaddress.ip_network(supernet_cidr, strict=False) for candidate in supernet.subnets(new_prefix=prefix_len): if any(candidate.overlaps(block) for block in occupied): continue return str(candidate) raise ValueError( f"No unused /{prefix_len} VPC CIDR found in region " f"({len(occupied)} existing block(s))." ) def validate_new_vpc_cidr_format(cidr: str) -> Optional[str]: """Return an error when ``cidr`` is not a valid AWS VPC IPv4 block.""" raw = (cidr or "").strip() if not raw: return "NEW_VPC_CIDR is required when creating a new VPC." try: network = ipaddress.ip_network(raw, strict=False) except ValueError: return f"NEW_VPC_CIDR={raw!r} is not a valid IPv4 CIDR block." if network.version != 4: return f"NEW_VPC_CIDR must be IPv4 (got {raw!r})." if network.prefixlen < 16 or network.prefixlen > 28: return ( f"NEW_VPC_CIDR={raw!r} must use a prefix length between /16 and /28 " f"(got /{network.prefixlen})." ) if not network.is_private: return ( f"NEW_VPC_CIDR={raw!r} should be in RFC1918 private space " "(10.0.0.0/8, 172.16.0.0/12, or 192.168.0.0/16)." ) return None def validate_new_vpc_cidr_no_overlap( cidr: str, existing_vpc_cidrs: Sequence[str], ) -> Optional[str]: """Return an error when ``cidr`` overlaps an existing VPC block in the region.""" format_error = validate_new_vpc_cidr_format(cidr) if format_error: return format_error candidate = ipaddress.ip_network(cidr.strip(), strict=False) for block in parse_ipv4_networks(existing_vpc_cidrs): if candidate.overlaps(block): return ( f"NEW_VPC_CIDR={candidate!s} overlaps existing VPC CIDR {block!s} " "in this account/region." ) return None def validate_new_vpc_cidr( cidr: str, existing_vpc_cidrs: Sequence[str], ) -> Optional[str]: """Validate format and regional non-overlap for a new VPC CIDR.""" return validate_new_vpc_cidr_no_overlap(cidr, existing_vpc_cidrs) def canonicalize_vpc_cidr(cidr: str) -> str: """Normalize a VPC CIDR string (e.g. 10.0.0.0/24).""" return str(ipaddress.ip_network(cidr.strip(), strict=False)) def subnet_cidr_prefix_len_for_tier(answers: "InstallAnswers", tier: str) -> int: """Return installer subnet sizing for public/private tiers and deployment mode.""" if tier == "public": if answers_use_public_subnets_only(answers): return EXPRESS_PUBLIC_SUBNET_CIDR_PREFIX_LEN return PUBLIC_SUBNET_CIDR_PREFIX_LEN return PRIVATE_SUBNET_CIDR_PREFIX_LEN def validate_public_subnet_cidr_for_express(cidr: str) -> Optional[str]: """Return an error when a public subnet is too small for Express managed ALB.""" raw = (cidr or "").strip() if not raw: return None try: network = ipaddress.ip_network(raw, strict=False) except ValueError: return f"Public subnet CIDR {raw!r} is not valid." if network.prefixlen > EXPRESS_ALB_MIN_SUBNET_PREFIX_LEN: return ( f"Public subnet {network!s} is too small for ECS Express Mode: the managed " f"ALB needs at least 8 free IP addresses per subnet. Use /" f"{EXPRESS_ALB_MIN_SUBNET_PREFIX_LEN} or larger (e.g. /27 in a /24 VPC)." ) return None def validate_new_vpc_cidr_env_values(values: Dict[str, str]) -> List[str]: """Live regional overlap check for NEW_VPC_CIDR before deploy.""" new_cidr = (values.get("NEW_VPC_CIDR") or "").strip() if not new_cidr or (values.get("VPC_NAME") or "").strip(): return [] region = (values.get("AWS_REGION") or "").strip() if not region: return ["AWS_REGION is required to validate NEW_VPC_CIDR."] occupied = list_vpc_cidr_blocks_in_region(region) err = validate_new_vpc_cidr(new_cidr, occupied) return [err] if err else [] def prompt_new_vpc_cidr(answers: "InstallAnswers", *, interactive: bool) -> None: """Fill ``answers.new_vpc_cidr`` via auto-suggest and/or manual wizard prompt.""" region = answers.aws_region occupied = list_vpc_cidr_blocks_in_region(region) if answers.new_vpc_cidr: err = validate_new_vpc_cidr(answers.new_vpc_cidr, occupied) if err: raise SystemExit(err) answers.new_vpc_cidr = canonicalize_vpc_cidr(answers.new_vpc_cidr) return auto_prompt = ( f"Auto-select lowest available /{VPC_CIDR_PREFIX_LEN} VPC CIDR " f"in {region}?" ) def _select_suggested() -> str: suggested = suggest_vpc_cidr_block(occupied) print(f"Selected VPC CIDR: {suggested}") return suggested def _prompt_manual(default_cidr: str) -> str: while True: raw = ask("New VPC CIDR", default_cidr) err = validate_new_vpc_cidr(raw, occupied) if err: print(err) continue return canonicalize_vpc_cidr(raw) if interactive: try: default_cidr = suggest_vpc_cidr_block(occupied) can_auto_select = True except ValueError as exc: print(f"Warning: {exc}") default_cidr = "" can_auto_select = False if can_auto_select and ask_yes_no(auto_prompt, default=True): answers.new_vpc_cidr = _select_suggested() else: if not default_cidr: print( f"Enter a non-overlapping RFC1918 VPC CIDR (/16–/28) for {region}." ) answers.new_vpc_cidr = _prompt_manual(default_cidr) return try: answers.new_vpc_cidr = _select_suggested() except ValueError as exc: raise SystemExit( f"Could not auto-select a VPC CIDR in {region}. " f"Pass --new-vpc-cidr with a non-overlapping RFC1918 /16–/28 block. {exc}" ) from exc def suggest_subnet_cidr_blocks( vpc_cidr: str, existing_subnet_cidrs: Sequence[str], count: int, *, prefix_len: int = SUBNET_CIDR_PREFIX_LEN, reserved_cidrs: Optional[Sequence[str]] = None, ) -> List[str]: """Return ``count`` lowest non-overlapping ``prefix_len`` blocks inside ``vpc_cidr``.""" if count < 1: return [] vpc_net = ipaddress.ip_network(vpc_cidr.strip(), strict=False) occupied = parse_ipv4_networks(existing_subnet_cidrs) if reserved_cidrs: occupied.extend(parse_ipv4_networks(reserved_cidrs)) suggestions: List[str] = [] for candidate in vpc_net.subnets(new_prefix=prefix_len): if any(candidate.overlaps(block) for block in occupied): continue suggestions.append(str(candidate)) occupied.append(candidate) if len(suggestions) >= count: return suggestions raise ValueError( f"Only {len(suggestions)} available /{prefix_len} subnet block(s) in " f"{vpc_cidr}; need {count}." ) def list_availability_zones(region: str) -> List[str]: import boto3 ec2 = boto3.client("ec2", region_name=region) zones = ec2.describe_availability_zones( Filters=[{"Name": "state", "Values": ["available"]}] ) return [z["ZoneName"] for z in zones.get("AvailabilityZones", [])] def list_subnets_in_vpc(vpc_id: str, region: str) -> List[Dict[str, str]]: sys.path.insert(0, str(CDK_DIR)) from cdk_functions import _get_existing_subnets_in_vpc data = _get_existing_subnets_in_vpc(vpc_id) subnets: List[Dict[str, str]] = [] for name, info in data.get("by_name", {}).items(): subnets.append( { "name": name, "id": info.get("id", ""), "cidr": info.get("cidr", ""), "az": info.get("az", ""), } ) return sorted(subnets, key=lambda x: x["name"]) def list_acm_certificates(region: str) -> List[Dict[str, str]]: import boto3 acm = boto3.client("acm", region_name=region) certs: List[Dict[str, str]] = [] paginator = acm.get_paginator("list_certificates") for page in paginator.paginate(CertificateStatuses=["ISSUED"]): for summary in page.get("CertificateSummaryList", []): arn = summary["CertificateArn"] detail = acm.describe_certificate(CertificateArn=arn)["Certificate"] domain = detail.get("DomainName", "") sans = detail.get("SubjectAlternativeNames", []) label = domain if sans and sans[0] != domain: label = f"{domain} (+{len(sans) - 1} SANs)" certs.append({"arn": arn, "domain": domain, "label": label}) return certs def list_igws_for_vpc(vpc_id: str, region: str) -> List[Dict[str, str]]: import boto3 ec2 = boto3.client("ec2", region_name=region) response = ec2.describe_internet_gateways( Filters=[{"Name": "attachment.vpc-id", "Values": [vpc_id]}] ) return [ {"id": igw["InternetGatewayId"]} for igw in response.get("InternetGateways", []) ] def list_albs_in_vpc(vpc_id: str, region: str) -> List[Dict[str, str]]: import boto3 elbv2 = boto3.client("elbv2", region_name=region) response = elbv2.describe_load_balancers() albs: List[Dict[str, str]] = [] for lb in response.get("LoadBalancers", []): if lb.get("VpcId") == vpc_id: albs.append( { "arn": lb["LoadBalancerArn"], "dns": lb["DNSName"], "name": lb["LoadBalancerName"], } ) return albs # --------------------------------------------------------------------------- # Env builder / writer # --------------------------------------------------------------------------- def format_list_env(values: Sequence[str], *, use_single_quotes: bool = False) -> str: if not values: return "[]" if use_single_quotes: inner = ", ".join(f"'{v}'" for v in values) else: inner = ", ".join(f'"{v}"' for v in values) return f"[{inner}]" SUBNET_TIER_MODES = ("auto", "existing", "create") def parse_subnet_name_csv(raw: str) -> List[str]: return [part.strip() for part in raw.split(",") if part.strip()] def resolve_subnet_tier_modes(args: argparse.Namespace) -> Tuple[str, str]: """CLI: --subnet-mode sets both tiers; per-tier flags override individually.""" default = getattr(args, "subnet_mode", None) or "auto" public = getattr(args, "public_subnet_mode", None) or default private = getattr(args, "private_subnet_mode", None) or default return public, private def apply_subnet_cli_flags(args: argparse.Namespace, answers: "InstallAnswers") -> None: if getattr(args, "public_subnet_names", None): answers.public_subnet_names = parse_subnet_name_csv(args.public_subnet_names) if getattr(args, "private_subnet_names", None): answers.private_subnet_names = parse_subnet_name_csv(args.private_subnet_names) def lookup_subnets_by_name( vpc_subnets: Sequence[Dict[str, str]], ) -> Dict[str, Dict[str, str]]: return {subnet["name"]: subnet for subnet in vpc_subnets if subnet.get("name")} def fill_existing_subnet_tier_metadata( names: Sequence[str], lookup: Dict[str, Dict[str, str]], *, cidrs_out: List[str], azs_out: List[str], ) -> List[str]: """Resolve CIDR/AZ for existing subnet names (order preserved).""" errors: List[str] = [] cidrs_out.clear() azs_out.clear() for name in names: info = lookup.get(name) if not info: errors.append(f"Subnet '{name}' was not found in the VPC.") continue cidr = (info.get("cidr") or "").strip() az = (info.get("az") or "").strip() if not az: errors.append(f"Subnet '{name}' has no availability zone in AWS.") continue cidrs_out.append(cidr) azs_out.append(az) return errors def enrich_existing_subnet_details_from_aws(answers: "InstallAnswers") -> List[str]: """Look up CIDR/AZ for tiers using existing named subnets (wizard or CLI).""" if answers.vpc_mode != "existing" or not answers.vpc_name: return [] tiers: List[Tuple[str, str, List[str], List[str], List[str]]] = [] if answers.public_subnet_mode == "existing" and answers.public_subnet_names: tiers.append( ( "Public", answers.public_subnet_mode, answers.public_subnet_names, answers.public_subnet_cidrs, answers.public_subnet_azs, ) ) if answers.private_subnet_mode == "existing" and answers.private_subnet_names: tiers.append( ( "Private", answers.private_subnet_mode, answers.private_subnet_names, answers.private_subnet_cidrs, answers.private_subnet_azs, ) ) if not tiers: return [] vpc_list = list_vpcs(answers.aws_region) vpc_id = next( (vpc["id"] for vpc in vpc_list if vpc["name"] == answers.vpc_name), None, ) if not vpc_id: return [f"VPC '{answers.vpc_name}' not found in {answers.aws_region}."] lookup = lookup_subnets_by_name(list_subnets_in_vpc(vpc_id, answers.aws_region)) errors: List[str] = [] for _label, _mode, names, cidrs_out, azs_out in tiers: if len(cidrs_out) == len(names) == len(azs_out) and names: continue errors.extend( fill_existing_subnet_tier_metadata( names, lookup, cidrs_out=cidrs_out, azs_out=azs_out, ) ) return errors def apply_subnet_tier_env( values: Dict[str, str], answers: "InstallAnswers", *, tier: str, mode: str, ) -> None: """Write PUBLIC_* or PRIVATE_* subnet keys for one tier (auto | existing | create).""" prefix = "PUBLIC" if tier == "public" else "PRIVATE" names = ( answers.public_subnet_names if tier == "public" else answers.private_subnet_names ) cidrs = ( answers.public_subnet_cidrs if tier == "public" else answers.private_subnet_cidrs ) azs = answers.public_subnet_azs if tier == "public" else answers.private_subnet_azs if mode == "auto": values[f"{prefix}_SUBNETS_TO_USE"] = "" values[f"{prefix}_SUBNET_CIDR_BLOCKS"] = "" values[f"{prefix}_SUBNET_AVAILABILITY_ZONES"] = "" elif mode == "existing": values[f"{prefix}_SUBNETS_TO_USE"] = format_list_env(names) if names and len(cidrs) == len(names) == len(azs): values[f"{prefix}_SUBNET_CIDR_BLOCKS"] = format_list_env( cidrs, use_single_quotes=True ) values[f"{prefix}_SUBNET_AVAILABILITY_ZONES"] = format_list_env( azs, use_single_quotes=True ) else: values[f"{prefix}_SUBNET_CIDR_BLOCKS"] = "" values[f"{prefix}_SUBNET_AVAILABILITY_ZONES"] = "" else: values[f"{prefix}_SUBNETS_TO_USE"] = format_list_env(names) values[f"{prefix}_SUBNET_CIDR_BLOCKS"] = format_list_env( cidrs, use_single_quotes=True ) values[f"{prefix}_SUBNET_AVAILABILITY_ZONES"] = format_list_env( azs, use_single_quotes=True ) def validate_subnet_answers(answers: "InstallAnswers") -> List[str]: errors: List[str] = [] if answers.vpc_mode != "existing": return errors tiers: List[tuple] = [ ( "Public", answers.public_subnet_mode, answers.public_subnet_names, answers.public_subnet_cidrs, ), ] if not answers_use_express_mode(answers) and not answers_use_public_subnets_only( answers ): tiers.append( ( "Private", answers.private_subnet_mode, answers.private_subnet_names, answers.private_subnet_cidrs, ) ) for label, mode, names, cidrs in tiers: if mode == "existing" and not names: errors.append( f"{label} subnets: provide at least one existing subnet name." ) if mode == "create" and not names: errors.append(f"{label} subnets: create mode requires subnet names.") if mode == "create" and names and cidrs and len(names) != len(cidrs): errors.append(f"{label} subnets: CIDR count must match subnet name count.") if label == "Public" and answers_use_express_mode(answers) and cidrs: for cidr in cidrs: subnet_error = validate_public_subnet_cidr_for_express(cidr) if subnet_error: errors.append(subnet_error) return errors def ask_subnet_tier_mode(tier_label: str) -> str: idx = ask_choice( f"{tier_label} subnets", [ "Auto-discover suitable subnets", "Use existing named subnets", "Create new stack-specific subnets (name + CIDR + AZ)", ], default_index=0, ) return SUBNET_TIER_MODES[idx] def configure_subnet_tier( answers: "InstallAnswers", tier: str, mode: str, vpc_subnets: List[Dict[str, str]], azs: Sequence[str], *, interactive: bool, vpc_cidr: str = "", reserved_subnet_cidrs: Optional[Sequence[str]] = None, ) -> None: """Collect subnet names/CIDRs for one tier in the wizard.""" is_public = tier == "public" label = "Public" if is_public else "Private" prefix_label = "Public" if is_public else "Private" if mode == "auto": return if mode == "existing": names = ( answers.public_subnet_names if is_public else answers.private_subnet_names ) if interactive and not names: print( f"Enter {label.lower()} subnet names (comma-separated), or pick from:" ) for subnet in vpc_subnets: print(f" - {subnet['name']} ({subnet['cidr']}, {subnet['az']})") names = parse_subnet_name_csv(ask(f"{label} subnet names", "")) if is_public: answers.public_subnet_names = names cidrs_out = answers.public_subnet_cidrs azs_out = answers.public_subnet_azs else: answers.private_subnet_names = names cidrs_out = answers.private_subnet_cidrs azs_out = answers.private_subnet_azs if names: lookup = lookup_subnets_by_name(vpc_subnets) missing = fill_existing_subnet_tier_metadata( names, lookup, cidrs_out=cidrs_out, azs_out=azs_out, ) for err in missing: print(f"Warning: {err}") return n_az = len(azs) prefix_label = "Public" if tier == "public" else "Private" prefix_len = subnet_cidr_prefix_len_for_tier(answers, tier) names = [f"{answers.cdk_prefix}{prefix_label}Subnet{i + 1}" for i in range(n_az)] if is_public: answers.public_subnet_names = names cidrs_list = answers.public_subnet_cidrs azs_list = answers.public_subnet_azs cidr_base = 0 else: answers.private_subnet_names = names cidrs_list = answers.private_subnet_cidrs azs_list = answers.private_subnet_azs cidr_base = 10 if interactive: print(f"Suggested AZs for {label.lower()} subnets: {', '.join(azs)}") auto_assign = ask_yes_no( f"Auto-assign lowest available /{prefix_len} CIDR blocks " f"for {label.lower()} subnets?", default=True, ) else: auto_assign = bool(vpc_cidr) existing_subnet_cidrs = [ subnet.get("cidr", "") for subnet in vpc_subnets if subnet.get("cidr") ] suggested: Optional[List[str]] = None if auto_assign and vpc_cidr: try: suggested = suggest_subnet_cidr_blocks( vpc_cidr, existing_subnet_cidrs, len(names), prefix_len=prefix_len, reserved_cidrs=reserved_subnet_cidrs, ) except ValueError as exc: print(f"Warning: {exc}") if not interactive: raise SystemExit(str(exc)) from exc cidrs_list.clear() azs_list.clear() for i, name in enumerate(names): if suggested: cidr = suggested[i] if interactive: print(f" {name}: {cidr}") elif interactive: default_cidr = f"10.0.{cidr_base + i}.0/{prefix_len}" if vpc_cidr: try: default_cidr = suggest_subnet_cidr_blocks( vpc_cidr, existing_subnet_cidrs + cidrs_list, 1, prefix_len=prefix_len, reserved_cidrs=reserved_subnet_cidrs, )[0] except ValueError: pass cidr = ask(f"CIDR for {label.lower()} {name}", default_cidr) else: cidr = f"10.0.{cidr_base + i}.0/{prefix_len}" cidrs_list.append(cidr) azs_list.append(azs[i % len(azs)]) @dataclass class InstallAnswers: profile: str = "demo" aws_account_id: str = "" aws_region: str = "" cdk_prefix: str = "" cognito_domain_prefix: str = "" s3_log_bucket_name: str = "" s3_output_bucket_name: str = "" github_branch: str = "main" vpc_mode: str = "existing" # new | existing vpc_name: str = "" new_vpc_cidr: str = "" public_subnet_mode: str = "auto" # auto | existing | create private_subnet_mode: str = "auto" # auto | existing | create public_subnet_names: List[str] = field(default_factory=list) private_subnet_names: List[str] = field(default_factory=list) public_subnet_cidrs: List[str] = field(default_factory=list) private_subnet_cidrs: List[str] = field(default_factory=list) public_subnet_azs: List[str] = field(default_factory=list) private_subnet_azs: List[str] = field(default_factory=list) existing_igw_id: str = "" existing_alb_arn: str = "" existing_alb_dns: str = "" acm_cert_arn: str = "" ssl_domain: str = "" cloudfront_geo: str = "" enable_pi_express: bool = False enable_pi_legacy: bool = False enable_service_connect: bool = False enable_s3_batch: bool = False enable_headless: bool = False enable_headless_output_notifications: bool = False headless_output_notify_email: str = "" headless_output_iam_user_name: str = "" enable_dynamo_usage_log_export: bool = False dynamo_export_schedule_time: str = "06:00" dynamo_export_schedule_days: str = "daily" dynamo_export_s3_key: str = "" dynamo_export_date_attribute: str = "timestamp" enable_bedrock_model_invocation_logging: bool = False bedrock_invocation_s3_prefix: str = "bedrock-logs" bedrock_invocation_log_group_name: str = "" bedrock_invocation_log_retention_days: str = "90" enable_appregistry: bool = False enable_application_resource_group: bool = True application_name: str = "" application_tag_key: str = "Application" ecs_memory: str = "8192" pi_alb_routing: str = "path" pi_alb_path_prefix: str = "/agent" pi_alb_host_header: str = "" pi_alb_listener_rule_priority: str = "" pi_gradio_port: str = "7862" sc_discovery_name: str = "llm-topic" pi_default_provider: str = "amazon-bedrock" write_pi_agent_env: bool = True overwrite_pi_agent_env: bool = False write_app_config_env: bool = True overwrite_app_config_env: bool = False custom_overrides: Dict[str, str] = field(default_factory=dict) python_path: Optional[str] = None @property def pi_enabled(self) -> bool: return self.enable_pi_express or self.enable_pi_legacy # Cognito hosted-UI prefix domains cannot contain these substrings (AWS docs). COGNITO_DOMAIN_RESERVED_SUBSTRINGS = ("amazon", "cognito", "aws") def sanitize_cognito_domain_prefix(raw: str, *, max_length: int = 63) -> str: """Normalize a Cognito domain prefix: allowed chars, no reserved words.""" prefix = re.sub(r"[^a-z0-9-]", "-", (raw or "").lower()) for reserved in COGNITO_DOMAIN_RESERVED_SUBSTRINGS: prefix = prefix.replace(reserved, "") prefix = re.sub(r"-{2,}", "-", prefix).strip("-") if max_length > 0: prefix = prefix[:max_length].strip("-") if not prefix or not re.match(r"^[a-z0-9]", prefix): return "llm-topic-app" prefix = prefix.rstrip("-") if not prefix: return "llm-topic-app" for reserved in COGNITO_DOMAIN_RESERVED_SUBSTRINGS: if reserved in prefix: return "llm-topic-app" return prefix def default_cognito_domain_prefix_from_cdk_prefix(cdk_prefix: str) -> str: """Derive a short installer default from CDK_PREFIX (globally unique hint).""" slug = re.sub(r"[^a-z0-9-]", "-", (cdk_prefix or "").lower()).strip("-") return sanitize_cognito_domain_prefix(slug, max_length=20) def validate_cognito_domain_prefix(prefix: str) -> Optional[str]: """Return an error message when prefix violates Cognito domain rules.""" cleaned = (prefix or "").strip().lower() if not cleaned: return "COGNITO_USER_POOL_DOMAIN_PREFIX is required." reserved = [word for word in COGNITO_DOMAIN_RESERVED_SUBSTRINGS if word in cleaned] if reserved: return ( "COGNITO_USER_POOL_DOMAIN_PREFIX cannot contain reserved words: " + ", ".join(reserved) + "." ) if not re.match(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$", cleaned): return ( "COGNITO_USER_POOL_DOMAIN_PREFIX must use lowercase letters, " "numbers, and hyphens only." ) return None def normalize_pi_path_prefix(raw: str) -> str: segment = (raw or "agent").strip().strip("/") return f"/{segment}" if segment else "/agent" def default_pi_listener_priority(use_cloudfront: bool = False) -> str: """Default Pi path/host rule priority (1–2 reserved for CloudFront / Express rules).""" del use_cloudfront # kept for call-site compatibility return "3" def derive_ecs_resource_names(cdk_prefix: str) -> Dict[str, str]: """ECS/CodeBuild resource names from CDK_PREFIX (matches cdk_config.py defaults).""" prefix = (cdk_prefix or "").strip() if not prefix: return {} return { "CLUSTER_NAME": f"{prefix}Cluster", "ECS_SERVICE_NAME": f"{prefix}ECSService", "ECS_EXPRESS_SERVICE_NAME": f"{prefix}ECSService", "ECS_PI_EXPRESS_SERVICE_NAME": f"{prefix}PiExpressService", "ECS_PI_SERVICE_NAME": f"{prefix}PiAgentService", "COGNITO_USER_POOL_CLIENT_SECRET_NAME": f"{prefix}ParamCognitoSecret", "CODEBUILD_PROJECT_NAME": f"{prefix}CodeBuildProject", "CODEBUILD_PI_PROJECT_NAME": f"{prefix}CodeBuildPiProject", } def derive_s3_bucket_names(cdk_prefix: str) -> Dict[str, str]: """S3 bucket names from CDK_PREFIX (must be written to cdk_config.env explicitly).""" prefix = (cdk_prefix or "").strip().lower() if not prefix: return {} return { "S3_LOG_CONFIG_BUCKET_NAME": f"{prefix}s3-logs", "S3_OUTPUT_BUCKET_NAME": f"{prefix}s3-output", } S3_BUCKET_NAME_MAX_LEN = 63 def normalize_s3_bucket_name(raw: str) -> str: """Lowercase S3 bucket name with allowed characters only (3–63 chars).""" name = re.sub(r"[^a-z0-9.-]", "-", (raw or "").lower()) name = re.sub(r"-{2,}", "-", name).strip("-") if len(name) < 3: name = f"{name}-bucket".strip("-") return name[:S3_BUCKET_NAME_MAX_LEN] def suggest_available_s3_bucket_name( preferred: str, account_id: str, *, s3_client: Any = None, max_attempts: int = 8, ) -> str: """Return a globally available S3 bucket name, preferring ``preferred``.""" from cdk_functions import resolve_s3_bucket_availability account = re.sub(r"[^0-9]", "", account_id or "") bases = [preferred] if account: bases.extend( [ f"{account}-{preferred}", f"{preferred}-{account}", ] ) seen: set[str] = set() for base in bases: candidate = normalize_s3_bucket_name(base) if not candidate or candidate in seen: continue seen.add(candidate) status, _ = resolve_s3_bucket_availability(candidate, s3_client=s3_client) if status == "available": return candidate for _ in range(max_attempts): suffix = secrets.token_hex(3) candidate = normalize_s3_bucket_name(f"{account}-{preferred}-{suffix}") if candidate in seen: continue seen.add(candidate) status, _ = resolve_s3_bucket_availability(candidate, s3_client=s3_client) if status == "available": return candidate raise RuntimeError( f"Could not find an available S3 bucket name near {preferred!r}. " "Set S3_LOG_CONFIG_BUCKET_NAME / S3_OUTPUT_BUCKET_NAME manually." ) def suggest_available_cognito_domain_prefix( preferred: str, account_id: str, region: str, *, cognito_client: Any = None, max_attempts: int = 8, ) -> str: """Return an available Cognito hosted UI domain prefix in ``region``.""" from cdk_functions import resolve_cognito_domain_prefix_availability account = re.sub(r"[^0-9]", "", account_id or "") preferred_clean = sanitize_cognito_domain_prefix(preferred, max_length=63) candidates = [preferred_clean] if account: candidates.extend( [ sanitize_cognito_domain_prefix( f"{account}-{preferred_clean}", max_length=63 ), sanitize_cognito_domain_prefix( f"{preferred_clean}-{account[-8:]}", max_length=63 ), ] ) seen: set[str] = set() for candidate in candidates: if not candidate or candidate in seen: continue seen.add(candidate) if ( resolve_cognito_domain_prefix_availability( candidate, region_name=region, cognito_client=cognito_client ) == "available" ): return candidate for _ in range(max_attempts): suffix = secrets.token_hex(2) candidate = sanitize_cognito_domain_prefix( f"{account}-{preferred_clean}-{suffix}", max_length=63 ) if not candidate or candidate in seen: continue seen.add(candidate) if ( resolve_cognito_domain_prefix_availability( candidate, region_name=region, cognito_client=cognito_client ) == "available" ): return candidate raise RuntimeError( f"Could not find an available Cognito domain prefix near {preferred!r}. " "Set COGNITO_USER_POOL_DOMAIN_PREFIX manually." ) def _prompt_globally_unique_s3_bucket( label: str, preferred: str, account_id: str, *, interactive: bool, assume_yes: bool, cli_override: str = "", s3_client: Any = None, ) -> str: from cdk_functions import resolve_s3_bucket_availability if cli_override: name = normalize_s3_bucket_name(cli_override) status, _ = resolve_s3_bucket_availability(name, s3_client=s3_client) if status == "globally_taken": raise SystemExit( f"{label}: bucket name {name!r} is taken globally by another AWS " "account. Choose a different --s3-log-bucket / --s3-output-bucket." ) return name preferred = normalize_s3_bucket_name(preferred) status, _ = resolve_s3_bucket_availability(preferred, s3_client=s3_client) if status == "owned": print(f"{label}: using existing bucket in this account: {preferred}") return preferred if status == "available": print(f"{label}: bucket name available: {preferred}") return preferred suggested = suggest_available_s3_bucket_name( preferred, account_id, s3_client=s3_client ) print( f"{label}: S3 bucket name {preferred!r} is taken globally by another AWS " f"account (S3 names are unique worldwide)." ) if not interactive: if assume_yes: print(f"{label}: using suggested name: {suggested}") return suggested raise SystemExit( f"{label}: bucket name {preferred!r} is taken globally. " f"Re-run with --yes to accept {suggested!r}, or pass an explicit bucket name." ) if ask_yes_no(f"Use suggested bucket name {suggested!r}?", default=True): return suggested while True: custom = ask(f"{label}: enter a globally unique bucket name", suggested) custom = normalize_s3_bucket_name(custom) custom_status, _ = resolve_s3_bucket_availability(custom, s3_client=s3_client) if custom_status == "available": return custom if custom_status == "owned": print(f"Bucket {custom!r} already exists in this account; will import it.") return custom print(f"Name {custom!r} is still taken globally. Try another name.") def _prompt_globally_unique_cognito_prefix( preferred: str, account_id: str, region: str, *, interactive: bool, assume_yes: bool, cli_override: str = "", cognito_client: Any = None, ) -> str: from cdk_functions import resolve_cognito_domain_prefix_availability if cli_override: override = sanitize_cognito_domain_prefix(cli_override) validation_error = validate_cognito_domain_prefix(override) if validation_error: raise SystemExit(validation_error) if ( resolve_cognito_domain_prefix_availability( override, region_name=region, cognito_client=cognito_client ) != "available" ): raise SystemExit( f"Cognito domain prefix {override!r} is not available in {region} " "(likely taken by another AWS account). Pass a different " "--cognito-prefix." ) return override preferred = sanitize_cognito_domain_prefix(preferred) availability = resolve_cognito_domain_prefix_availability( preferred, region_name=region, cognito_client=cognito_client ) if availability == "taken": suggested = suggest_available_cognito_domain_prefix( preferred, account_id, region, cognito_client=cognito_client ) print( f"Cognito hosted UI domain prefix {preferred!r} is not available in " f"{region} (likely taken by another AWS account)." ) if not interactive: if assume_yes: print(f"Using suggested Cognito domain prefix: {suggested}") return suggested raise SystemExit( f"Cognito domain prefix {preferred!r} is not available in {region}. " f"Re-run with --yes to accept {suggested!r}, or pass --cognito-prefix." ) if ask_yes_no( f"Use suggested Cognito domain prefix {suggested!r}?", default=True ): return suggested preferred = suggested if not interactive: print(f"Cognito domain prefix available: {preferred}") return preferred while True: raw = ask( "Cognito hosted UI domain prefix " "(must be globally unique in this region; " "cannot contain aws, amazon, or cognito)", preferred, ) choice = sanitize_cognito_domain_prefix(raw) validation_error = validate_cognito_domain_prefix(choice) if validation_error: print(validation_error) continue if ( resolve_cognito_domain_prefix_availability( choice, region_name=region, cognito_client=cognito_client ) == "available" ): if choice != preferred: print(f"Using Cognito domain prefix: {choice}") else: print(f"Cognito domain prefix available: {choice}") return choice print( f"Prefix {choice!r} is not available in {region} " "(likely taken by another AWS account). Try another prefix." ) def resolve_globally_unique_install_names( answers: InstallAnswers, *, interactive: bool, assume_yes: bool, args: Optional[argparse.Namespace] = None, ) -> None: """Check S3 bucket and Cognito domain names; prompt for alternatives when taken.""" import boto3 args = args or argparse.Namespace() region = answers.aws_region account_id = answers.aws_account_id s3_client = boto3.client("s3", region_name=region) cognito_client = boto3.client("cognito-idp", region_name=region) defaults = derive_s3_bucket_names(answers.cdk_prefix) answers.s3_log_bucket_name = _prompt_globally_unique_s3_bucket( "S3 log/config bucket", defaults.get("S3_LOG_CONFIG_BUCKET_NAME", ""), account_id, interactive=interactive, assume_yes=assume_yes, cli_override=getattr(args, "s3_log_bucket", "") or "", s3_client=s3_client, ) answers.s3_output_bucket_name = _prompt_globally_unique_s3_bucket( "S3 output bucket", defaults.get("S3_OUTPUT_BUCKET_NAME", ""), account_id, interactive=interactive, assume_yes=assume_yes, cli_override=getattr(args, "s3_output_bucket", "") or "", s3_client=s3_client, ) if answers_use_headless(answers): answers.cognito_domain_prefix = "" return preferred_cognito = default_cognito_domain_prefix_from_cdk_prefix( answers.cdk_prefix ) answers.cognito_domain_prefix = _prompt_globally_unique_cognito_prefix( preferred_cognito, account_id, region, interactive=interactive, assume_yes=assume_yes, cli_override=getattr(args, "cognito_prefix", "") or "", cognito_client=cognito_client, ) def validate_globally_unique_env_values(values: Dict[str, str]) -> List[str]: """Live AWS checks for globally unique resource names before deploy.""" from cdk_functions import ( resolve_cognito_domain_prefix_availability, resolve_s3_bucket_availability, ) errors: List[str] = [] region = (values.get("AWS_REGION") or "").strip() for env_key in ("S3_LOG_CONFIG_BUCKET_NAME", "S3_OUTPUT_BUCKET_NAME"): bucket_name = (values.get(env_key) or "").strip().lower() if not bucket_name: continue status, _ = resolve_s3_bucket_availability(bucket_name) if status == "globally_taken": errors.append( f"{env_key}={bucket_name!r} is taken globally by another AWS account. " "Re-run cdk_install.py to pick a unique name." ) if values.get("ENABLE_HEADLESS_DEPLOYMENT") != "True": cognito_prefix = (values.get("COGNITO_USER_POOL_DOMAIN_PREFIX") or "").strip() if cognito_prefix and region: if ( resolve_cognito_domain_prefix_availability( cognito_prefix, region_name=region ) == "taken" ): errors.append( f"COGNITO_USER_POOL_DOMAIN_PREFIX={cognito_prefix!r} is not " f"available in {region} (taken by another AWS account or existing " "pool). Re-run cdk_install.py to pick a unique prefix." ) return errors def resolve_fixup_env_values(values: Dict[str, str]) -> Dict[str, str]: """Fill missing ECS cluster/service keys from CDK_PREFIX for post-deploy boto3 calls.""" resolved = dict(values) for key, default in derive_ecs_resource_names( resolved.get("CDK_PREFIX", "") ).items(): if not (resolved.get(key) or "").strip(): resolved[key] = default for key, default in derive_s3_bucket_names(resolved.get("CDK_PREFIX", "")).items(): if not (resolved.get(key) or "").strip(): resolved[key] = default return resolved def merge_preset( profile: str, overrides: Optional[Dict[str, str]] = None ) -> Dict[str, str]: base: Dict[str, str] = {} if profile == "demo": base.update(DEMO_PRESET) elif profile == "production": base.update(PRODUCTION_PRESET) elif profile == "headless": base.update(HEADLESS_PRESET) if overrides: base.update(overrides) return base def answers_preset_profile(answers: "InstallAnswers") -> str: """Preset profile for env defaults (legacy ``headless`` maps to demo).""" if answers.profile == "headless": return "demo" return answers.profile def answers_use_headless(answers: "InstallAnswers") -> bool: return answers.profile == "headless" or answers.enable_headless def profile_allows_headless_add_on(answers: "InstallAnswers") -> bool: """Headless batch is incompatible with the Demonstration (Express) route.""" if answers.profile == "demo": return False if answers.profile in ("production", "headless"): return True if answers.profile == "custom": return answers.custom_overrides.get("USE_ECS_EXPRESS_MODE") != "True" return False def headless_profile_error(answers: "InstallAnswers") -> Optional[str]: if not answers_use_headless(answers): return None if answers.profile == "headless": return None if answers.profile == "demo": return ( "Headless batch mode is not available with the Demonstration (Express) " "profile. Use the Headless shortcut profile, Production with headless, " "or Custom without ECS Express." ) if ( answers.profile == "custom" and answers.custom_overrides.get("USE_ECS_EXPRESS_MODE") == "True" ): return ( "Headless batch mode requires USE_ECS_EXPRESS_MODE=False. " "Disable ECS Express in the custom profile or omit --headless." ) return None def validate_notify_email(email: str) -> Optional[str]: """Return an error message when email is not a plausible notification address.""" cleaned = (email or "").strip() if not cleaned: return "Notification email address is required." if not re.match(r"^[^@\s]+@[^@\s]+\.[^@\s]+$", cleaned): return f"Invalid notification email address: {cleaned!r}" return None DYNAMO_EXPORT_SCHEDULE_DAY_PRESETS = { "daily": "*", "weekdays": "MON-FRI", "weekends": "SAT-SUN", } def default_dynamo_export_s3_key() -> str: return "reports/dynamodb-usage/dynamodb_logs_export.csv" def parse_schedule_time_hhmm(raw: str) -> Tuple[int, int]: match = re.match(r"^(\d{1,2}):(\d{2})$", (raw or "").strip()) if not match: raise ValueError( "Schedule time must use HH:MM in 24-hour UTC format, e.g. 06:00." ) hour, minute = int(match.group(1)), int(match.group(2)) if hour > 23 or minute > 59: raise ValueError("Schedule time hour must be 0-23 and minute 0-59.") return hour, minute def validate_schedule_time_hhmm(raw: str) -> Optional[str]: try: parse_schedule_time_hhmm(raw) return None except ValueError as exc: return str(exc) def build_dynamo_export_cron_expression( hour: int, minute: int, days_preset: str ) -> str: """Build an EventBridge cron expression (UTC).""" dow = DYNAMO_EXPORT_SCHEDULE_DAY_PRESETS.get( (days_preset or "").strip().lower(), (days_preset or "*").strip().upper() ) return f"cron({minute} {hour} ? * {dow} *)" def prompt_dynamo_usage_log_export_options( answers: "InstallAnswers", args: argparse.Namespace, *, interactive: bool, ) -> None: if getattr(args, "dynamo_usage_log_export", False): answers.enable_dynamo_usage_log_export = True elif getattr(args, "no_dynamo_usage_log_export", False): answers.enable_dynamo_usage_log_export = False elif interactive: answers.enable_dynamo_usage_log_export = ask_yes_no( "Enable scheduled DynamoDB usage log export to S3 (EventBridge)?", default=False, ) else: return if not answers.enable_dynamo_usage_log_export: return if getattr(args, "dynamo_export_schedule_time", "").strip(): answers.dynamo_export_schedule_time = args.dynamo_export_schedule_time.strip() elif interactive: while True: raw = ask( "Export schedule time (UTC, HH:MM)", answers.dynamo_export_schedule_time, ) err = validate_schedule_time_hhmm(raw) if not err: answers.dynamo_export_schedule_time = raw break print(err) if getattr(args, "dynamo_export_schedule_days", "").strip(): answers.dynamo_export_schedule_days = ( args.dynamo_export_schedule_days.strip().lower() ) elif interactive: idx = ask_choice( "Run export on which days?", ["Every day", "Weekdays (Mon-Fri)", "Weekends (Sat-Sun)"], default_index=0, ) answers.dynamo_export_schedule_days = ["daily", "weekdays", "weekends"][idx] default_key = answers.dynamo_export_s3_key or default_dynamo_export_s3_key() if getattr(args, "dynamo_export_s3_key", "").strip(): answers.dynamo_export_s3_key = args.dynamo_export_s3_key.strip() elif interactive: answers.dynamo_export_s3_key = ( ask("S3 object key for the CSV export", default_key).strip() or default_key ) if getattr(args, "dynamo_export_date_attribute", "").strip(): answers.dynamo_export_date_attribute = args.dynamo_export_date_attribute.strip() elif interactive: raw = ask( "DynamoDB date attribute for filtering", answers.dynamo_export_date_attribute, ) if raw.strip(): answers.dynamo_export_date_attribute = raw.strip() def default_bedrock_invocation_log_group_name(cdk_prefix: str) -> str: """Default CloudWatch log group for Bedrock model invocation logging.""" prefix = (cdk_prefix or "").strip() return f"/aws/bedrock/{prefix}model-invocations".rstrip("/") def prompt_bedrock_model_invocation_logging_options( answers: "InstallAnswers", args: argparse.Namespace, *, interactive: bool, ) -> None: """ Configure account/Region Bedrock model invocation logging to S3 + CloudWatch. Writes ENABLE_BEDROCK_MODEL_INVOCATION_LOGGING and related keys into cdk_config.env. """ if getattr(args, "bedrock_model_invocation_logging", False): answers.enable_bedrock_model_invocation_logging = True elif getattr(args, "no_bedrock_model_invocation_logging", False): answers.enable_bedrock_model_invocation_logging = False elif interactive: default_on = answers_use_headless(answers) answers.enable_bedrock_model_invocation_logging = ask_yes_no( "Enable Amazon Bedrock model invocation logging " "(S3 log bucket + CloudWatch; account/Region-level setting)?", default=default_on, ) else: return if not answers.enable_bedrock_model_invocation_logging: return default_prefix = answers.bedrock_invocation_s3_prefix or "bedrock-logs" if getattr(args, "bedrock_invocation_s3_prefix", "").strip(): answers.bedrock_invocation_s3_prefix = ( args.bedrock_invocation_s3_prefix.strip().strip("/") ) elif interactive: answers.bedrock_invocation_s3_prefix = ( ask( "S3 key prefix for Bedrock invocation logs (on the log/config bucket)", default_prefix, ) .strip() .strip("/") or default_prefix ) else: answers.bedrock_invocation_s3_prefix = default_prefix default_log_group = ( answers.bedrock_invocation_log_group_name.strip() or default_bedrock_invocation_log_group_name(answers.cdk_prefix) ) if getattr(args, "bedrock_invocation_log_group", "").strip(): answers.bedrock_invocation_log_group_name = ( args.bedrock_invocation_log_group.strip() ) elif interactive: answers.bedrock_invocation_log_group_name = ( ask( "CloudWatch log group for Bedrock invocation logs", default_log_group ).strip() or default_log_group ) else: answers.bedrock_invocation_log_group_name = default_log_group default_retention = answers.bedrock_invocation_log_retention_days or "90" if getattr(args, "bedrock_invocation_log_retention_days", None) is not None: answers.bedrock_invocation_log_retention_days = str( args.bedrock_invocation_log_retention_days ).strip() elif interactive: while True: raw = ask( "CloudWatch log retention (days)", default_retention, ).strip() try: days = int(raw) if days <= 0: raise ValueError answers.bedrock_invocation_log_retention_days = str(days) break except ValueError: print("Enter a positive whole number of days (e.g. 90).") else: answers.bedrock_invocation_log_retention_days = default_retention def default_application_name(cdk_prefix: str) -> str: """Default Application tag value / Resource Group identity from CDK_PREFIX.""" return f"{(cdk_prefix or '').strip()}llm-topic-modeller" def prompt_application_tagging_and_appregistry_options( answers: "InstallAnswers", args: argparse.Namespace, *, interactive: bool, ) -> None: """ Configure application tags / Resource Groups (preferred) and legacy AppRegistry. AppRegistry / myApplications closes to new customers after 30 Jul 2026. """ if getattr(args, "appregistry", False): answers.enable_appregistry = True elif getattr(args, "no_appregistry", False): answers.enable_appregistry = False elif interactive: answers.enable_appregistry = ask_yes_no( "Enable legacy AppRegistry / myApplications stack " "(opt-in; new customers cut off after 30 Jul 2026)?", default=False, ) # else: keep InstallAnswers default (False) if getattr(args, "application_resource_group", False): answers.enable_application_resource_group = True elif getattr(args, "no_application_resource_group", False): answers.enable_application_resource_group = False elif interactive: answers.enable_application_resource_group = ask_yes_no( "Tag all CDK-managed resources and create a Resource Group for this app " "(recommended for monitoring without AppRegistry)?", default=True, ) # else: keep default True if not answers.enable_application_resource_group: return default_tag_key = (answers.application_tag_key or "Application").strip() or ( "Application" ) if getattr(args, "application_tag_key", "").strip(): answers.application_tag_key = args.application_tag_key.strip() elif interactive: answers.application_tag_key = ( ask("Application tag key", default_tag_key).strip() or default_tag_key ) else: answers.application_tag_key = default_tag_key default_name = answers.application_name.strip() or default_application_name( answers.cdk_prefix ) if getattr(args, "application_name", "").strip(): answers.application_name = args.application_name.strip() elif interactive: answers.application_name = ( ask( "Application name (tag value / Resource Group filter)", default_name ).strip() or default_name ) else: answers.application_name = default_name def validate_install_answers(answers: "InstallAnswers") -> List[str]: errors: List[str] = [] msg = headless_profile_error(answers) if msg: errors.append(msg) if answers.enable_headless_output_notifications: if not answers_use_headless(answers): errors.append( "Headless output notifications require headless batch deployment." ) email_error = validate_notify_email(answers.headless_output_notify_email) if email_error: errors.append(email_error) if answers.enable_dynamo_usage_log_export: time_error = validate_schedule_time_hhmm(answers.dynamo_export_schedule_time) if time_error: errors.append(time_error) days = (answers.dynamo_export_schedule_days or "").strip().lower() if days not in DYNAMO_EXPORT_SCHEDULE_DAY_PRESETS: errors.append( "DynamoDB export schedule days must be one of: " + ", ".join(DYNAMO_EXPORT_SCHEDULE_DAY_PRESETS) ) if not (answers.dynamo_export_s3_key.strip() or default_dynamo_export_s3_key()): errors.append("DynamoDB export S3 object key is required.") if answers.enable_bedrock_model_invocation_logging: if not (answers.bedrock_invocation_s3_prefix or "").strip(): errors.append( "Bedrock model invocation logging requires an S3 key prefix " "(bedrock_invocation_s3_prefix)." ) log_group = (answers.bedrock_invocation_log_group_name or "").strip() if not log_group: errors.append( "Bedrock model invocation logging requires a CloudWatch log group name." ) try: retention = int( (answers.bedrock_invocation_log_retention_days or "90").strip() ) if retention <= 0: raise ValueError except ValueError: errors.append( "Bedrock invocation log retention days must be a positive integer." ) if answers.enable_application_resource_group: app_name = answers.application_name.strip() or default_application_name( answers.cdk_prefix ) if not app_name: errors.append( "Application Resource Group requires an application name " "(APPLICATION_NAME)." ) if not (answers.application_tag_key or "").strip(): errors.append("Application Resource Group requires an application tag key.") return errors def answers_use_express_mode(answers: "InstallAnswers") -> bool: """True when the selected install profile/options enable ECS Express ingress.""" if answers_use_headless(answers): return False preset = merge_preset(answers.profile, answers.custom_overrides) return preset.get("USE_ECS_EXPRESS_MODE") == "True" def answers_use_public_subnets_only(answers: "InstallAnswers") -> bool: """Express ingress or demo-style headless batch (public subnets, no private install).""" if answers_use_express_mode(answers): return True if not answers_use_headless(answers): return False return answers_preset_profile(answers) in ("headless", "demo", "custom") def build_env_values(answers: InstallAnswers) -> Dict[str, str]: cdk_folder = str(CDK_DIR).replace("\\", "/") if not cdk_folder.endswith("/"): cdk_folder += "/" s3_bucket_names = derive_s3_bucket_names(answers.cdk_prefix) if answers.s3_log_bucket_name: s3_bucket_names["S3_LOG_CONFIG_BUCKET_NAME"] = answers.s3_log_bucket_name if answers.s3_output_bucket_name: s3_bucket_names["S3_OUTPUT_BUCKET_NAME"] = answers.s3_output_bucket_name values: Dict[str, str] = merge_preset( answers_preset_profile(answers), answers.custom_overrides ) values.update( { "CDK_PREFIX": answers.cdk_prefix, "AWS_REGION": answers.aws_region, "AWS_ACCOUNT_ID": answers.aws_account_id, "CDK_FOLDER": cdk_folder, **derive_ecs_resource_names(answers.cdk_prefix), **s3_bucket_names, "CONTEXT_FILE": "precheck.context.json", "COGNITO_USER_POOL_DOMAIN_PREFIX": sanitize_cognito_domain_prefix( answers.cognito_domain_prefix ), "GITHUB_REPO_BRANCH": answers.github_branch, "ECS_TASK_MEMORY_SIZE": answers.ecs_memory, "EXISTING_IGW_ID": answers.existing_igw_id, "EXISTING_LOAD_BALANCER_ARN": answers.existing_alb_arn, "EXISTING_LOAD_BALANCER_DNS": answers.existing_alb_dns or "placeholder_load_balancer_dns.net", "ENABLE_PI_AGENT_EXPRESS_SERVICE": ( "True" if answers.enable_pi_express else "False" ), "ENABLE_PI_AGENT_ECS_SERVICE": ( "True" if answers.enable_pi_legacy else "False" ), "ENABLE_ECS_SERVICE_CONNECT": ( "True" if answers.enable_service_connect else "False" ), "ENABLE_S3_BATCH_ECS_TRIGGER": ( "True" if answers.enable_s3_batch or answers_use_headless(answers) else "False" ), "ENABLE_HEADLESS_DEPLOYMENT": ( "True" if answers_use_headless(answers) else "False" ), } ) if answers_use_headless(answers): values.update( { "ENABLE_HEADLESS_DEPLOYMENT": "True", "ENABLE_S3_BATCH_ECS_TRIGGER": "True", "USE_ECS_EXPRESS_MODE": "False", "USE_CLOUDFRONT": "False", "RUN_USEAST_STACK": "False", "COGNITO_AUTH": "False", "ENABLE_PI_AGENT_EXPRESS_SERVICE": "False", "ENABLE_PI_AGENT_ECS_SERVICE": "False", "ENABLE_ECS_SERVICE_CONNECT": "False", } ) if answers.enable_headless_output_notifications: iam_user = ( answers.headless_output_iam_user_name.strip() or f"{answers.cdk_prefix}s3-output-reader" ) values.update( { "ENABLE_HEADLESS_OUTPUT_NOTIFICATIONS": "True", "HEADLESS_OUTPUT_NOTIFY_EMAIL": ( answers.headless_output_notify_email.strip() ), "HEADLESS_OUTPUT_IAM_USER_NAME": iam_user, } ) if answers.enable_dynamo_usage_log_export: hour, minute = parse_schedule_time_hhmm(answers.dynamo_export_schedule_time) s3_key = answers.dynamo_export_s3_key.strip() or default_dynamo_export_s3_key() values.update( { "ENABLE_DYNAMODB_USAGE_LOG_EXPORT": "True", "DYNAMODB_USAGE_LOG_EXPORT_SCHEDULE": build_dynamo_export_cron_expression( hour, minute, answers.dynamo_export_schedule_days ), "DYNAMODB_USAGE_LOG_EXPORT_S3_KEY": s3_key, "DYNAMODB_USAGE_LOG_EXPORT_DATE_ATTRIBUTE": ( answers.dynamo_export_date_attribute.strip() or "timestamp" ), } ) if answers.enable_bedrock_model_invocation_logging: values.update( { "ENABLE_BEDROCK_MODEL_INVOCATION_LOGGING": "True", "BEDROCK_MODEL_INVOCATION_S3_PREFIX": ( answers.bedrock_invocation_s3_prefix.strip().strip("/") or "bedrock-logs" ), "BEDROCK_MODEL_INVOCATION_LOG_GROUP_NAME": ( answers.bedrock_invocation_log_group_name.strip() or default_bedrock_invocation_log_group_name(answers.cdk_prefix) ), "BEDROCK_MODEL_INVOCATION_LOG_RETENTION_DAYS": ( answers.bedrock_invocation_log_retention_days.strip() or "90" ), } ) # Application tags + Resource Groups (preferred); AppRegistry is legacy opt-in. # Always write tagging identity so Tags.of(app) has stable values even when # Resource Group creation is disabled. app_name = answers.application_name.strip() or default_application_name( answers.cdk_prefix ) tag_key = (answers.application_tag_key or "Application").strip() or "Application" values["APPLICATION_NAME"] = app_name values["APPLICATION_TAG_KEY"] = tag_key values["APPLICATION_RESOURCE_GROUP_NAME"] = f"{app_name}-resources" values["ENABLE_APPLICATION_RESOURCE_GROUP"] = ( "True" if answers.enable_application_resource_group else "False" ) values["ENABLE_APPREGISTRY"] = "True" if answers.enable_appregistry else "False" if answers.enable_appregistry: values["APPREGISTRY_STACK_NAME"] = ( f"{answers.cdk_prefix}{APPREGISTRY_STACK_SUFFIX}" ) values["APPREGISTRY_APPLICATION_NAME"] = app_name use_cloudfront = values.get("USE_CLOUDFRONT") == "True" if answers.pi_enabled: values.update( { "PI_GRADIO_PORT": answers.pi_gradio_port, "ECS_SERVICE_CONNECT_DISCOVERY_NAME": answers.sc_discovery_name, } ) if answers.enable_pi_express: values["ECS_EXPRESS_SC_PORT_NAME"] = "port-7860" values["ECS_PI_EXPRESS_SC_PORT_NAME"] = f"port-{answers.pi_gradio_port}" else: routing = answers.pi_alb_routing.strip().lower() path_prefix = normalize_pi_path_prefix(answers.pi_alb_path_prefix) priority = ( answers.pi_alb_listener_rule_priority.strip() or default_pi_listener_priority(use_cloudfront) ) values.update( { "PI_ALB_ROUTING": routing, "PI_ALB_PATH_PREFIX": path_prefix, "PI_ALB_LISTENER_RULE_PRIORITY": priority, } ) if routing in ("host", "both"): values["PI_ALB_HOST_HEADER"] = answers.pi_alb_host_header.strip() if answers.profile == "production": values["ACM_SSL_CERTIFICATE_ARN"] = answers.acm_cert_arn values["SSL_CERTIFICATE_DOMAIN"] = answers.ssl_domain values["CLOUDFRONT_DOMAIN"] = "cloudfront_placeholder.net" if answers.cloudfront_geo: values["CLOUDFRONT_GEO_RESTRICTION"] = answers.cloudfront_geo if answers.vpc_mode == "new": values["VPC_NAME"] = "" values["NEW_VPC_CIDR"] = answers.new_vpc_cidr values["NEW_VPC_DEFAULT_NAME"] = f"{answers.cdk_prefix}vpc" values["PUBLIC_SUBNETS_TO_USE"] = "" values["PRIVATE_SUBNETS_TO_USE"] = "" values["PUBLIC_SUBNET_CIDR_BLOCKS"] = "" values["PRIVATE_SUBNET_CIDR_BLOCKS"] = "" values["PUBLIC_SUBNET_AVAILABILITY_ZONES"] = "" values["PRIVATE_SUBNET_AVAILABILITY_ZONES"] = "" else: values["VPC_NAME"] = answers.vpc_name values["NEW_VPC_CIDR"] = "" apply_subnet_tier_env( values, answers, tier="public", mode=answers.public_subnet_mode ) if answers_use_public_subnets_only(answers): values["ECS_EXPRESS_USE_PUBLIC_SUBNETS"] = "True" values["PRIVATE_SUBNETS_TO_USE"] = "" values["PRIVATE_SUBNET_CIDR_BLOCKS"] = "" values["PRIVATE_SUBNET_AVAILABILITY_ZONES"] = "" else: apply_subnet_tier_env( values, answers, tier="private", mode=answers.private_subnet_mode ) # Pi agent is not supported in llm_topic_modeller yet (stack code retained for future). values["ENABLE_PI_AGENT_EXPRESS_SERVICE"] = "False" values["ENABLE_PI_AGENT_ECS_SERVICE"] = "False" return values def validate_env_values(values: Dict[str, str]) -> List[str]: errors: List[str] = [] express = values.get("USE_ECS_EXPRESS_MODE") == "True" acm = (values.get("ACM_SSL_CERTIFICATE_ARN") or "").strip() if express and acm: errors.append( "USE_ECS_EXPRESS_MODE=True cannot be used with ACM_SSL_CERTIFICATE_ARN." ) if values.get("ENABLE_ECS_SERVICE_CONNECT") == "True" and express: errors.append( "ENABLE_ECS_SERVICE_CONNECT requires legacy Fargate (not Express)." ) if values.get("ENABLE_S3_BATCH_ECS_TRIGGER") == "True" and express: errors.append( "ENABLE_S3_BATCH_ECS_TRIGGER requires legacy Fargate (not Express)." ) headless = values.get("ENABLE_HEADLESS_DEPLOYMENT") == "True" if headless: if values.get("ENABLE_S3_BATCH_ECS_TRIGGER") != "True": errors.append( "ENABLE_HEADLESS_DEPLOYMENT requires ENABLE_S3_BATCH_ECS_TRIGGER=True." ) if express: errors.append( "ENABLE_HEADLESS_DEPLOYMENT cannot be combined with USE_ECS_EXPRESS_MODE=True " "(use --profile headless, production --headless, or custom without Express)." ) if values.get("USE_CLOUDFRONT") == "True": errors.append( "ENABLE_HEADLESS_DEPLOYMENT is incompatible with USE_CLOUDFRONT=True." ) pi_ecs = values.get("ENABLE_PI_AGENT_ECS_SERVICE") == "True" pi_express = values.get("ENABLE_PI_AGENT_EXPRESS_SERVICE") == "True" if pi_ecs and pi_express: errors.append("Enable at most one agent mode (ECS or Express).") if pi_express and not express: errors.append( "ENABLE_PI_AGENT_EXPRESS_SERVICE requires USE_ECS_EXPRESS_MODE=True." ) if pi_ecs and express: errors.append( "ENABLE_PI_AGENT_ECS_SERVICE requires USE_ECS_EXPRESS_MODE=False." ) vpc_name = (values.get("VPC_NAME") or "").strip() new_cidr = (values.get("NEW_VPC_CIDR") or "").strip() if not vpc_name and not new_cidr: errors.append("Set VPC_NAME (existing VPC) or NEW_VPC_CIDR (new VPC).") elif new_cidr and not vpc_name: format_error = validate_new_vpc_cidr_format(new_cidr) if format_error: errors.append(format_error) if express: public_cidrs_raw = (values.get("PUBLIC_SUBNET_CIDR_BLOCKS") or "").strip() if public_cidrs_raw: try: import ast parsed = ast.literal_eval(public_cidrs_raw) public_cidrs = parsed if isinstance(parsed, list) else [str(parsed)] except (SyntaxError, ValueError): public_cidrs = [ part.strip().strip("'\"") for part in public_cidrs_raw.split(",") if part.strip() ] for cidr in public_cidrs: subnet_error = validate_public_subnet_cidr_for_express(str(cidr)) if subnet_error: errors.append(subnet_error) if values.get("USE_CLOUDFRONT") == "True" and not express: if not acm: errors.append( "Production CloudFront path requires ACM_SSL_CERTIFICATE_ARN." ) if not (values.get("SSL_CERTIFICATE_DOMAIN") or "").strip(): errors.append("Production CloudFront path requires SSL_CERTIFICATE_DOMAIN.") if not (values.get("CDK_PREFIX") or "").strip(): errors.append("CDK_PREFIX is required.") prefix_lower = (values.get("CDK_PREFIX") or "").strip().lower() for env_key, bare_suffix in ( ("S3_LOG_CONFIG_BUCKET_NAME", "s3-logs"), ("S3_OUTPUT_BUCKET_NAME", "s3-output"), ): bucket_name = (values.get(env_key) or "").strip().lower() if prefix_lower and bucket_name == bare_suffix: errors.append( f"{env_key} must include CDK_PREFIX (got bare '{bare_suffix}'; " f"expected '{prefix_lower}{bare_suffix}'). Re-run cdk_install or set " f"{env_key}={prefix_lower}{bare_suffix} in config/cdk_config.env." ) if values.get("ENABLE_HEADLESS_DEPLOYMENT") != "True": cognito_prefix_error = validate_cognito_domain_prefix( values.get("COGNITO_USER_POOL_DOMAIN_PREFIX", "") ) if cognito_prefix_error: errors.append(cognito_prefix_error) if pi_ecs and values.get("ENABLE_ECS_SERVICE_CONNECT") != "True": errors.append( "ENABLE_PI_AGENT_ECS_SERVICE=True requires ENABLE_ECS_SERVICE_CONNECT=True." ) if headless and ( pi_ecs or pi_express or values.get("ENABLE_ECS_SERVICE_CONNECT") == "True" ): errors.append( "ENABLE_HEADLESS_DEPLOYMENT is incompatible with agent mode or Service Connect." ) if values.get("ENABLE_HEADLESS_OUTPUT_NOTIFICATIONS") == "True": if not headless: errors.append( "ENABLE_HEADLESS_OUTPUT_NOTIFICATIONS requires " "ENABLE_HEADLESS_DEPLOYMENT=True." ) email_error = validate_notify_email( values.get("HEADLESS_OUTPUT_NOTIFY_EMAIL", "") ) if email_error: errors.append(email_error) if values.get("ENABLE_DYNAMODB_USAGE_LOG_EXPORT") == "True": if values.get("SAVE_LOGS_TO_DYNAMODB") != "True": errors.append( "ENABLE_DYNAMODB_USAGE_LOG_EXPORT requires SAVE_LOGS_TO_DYNAMODB=True." ) schedule = (values.get("DYNAMODB_USAGE_LOG_EXPORT_SCHEDULE") or "").strip() if not schedule.startswith("cron(") and not schedule.startswith("rate("): errors.append( "DYNAMODB_USAGE_LOG_EXPORT_SCHEDULE must be an EventBridge " f"cron(...) or rate(...) expression; got {schedule!r}." ) if not (values.get("DYNAMODB_USAGE_LOG_EXPORT_S3_KEY") or "").strip(): errors.append( "DYNAMODB_USAGE_LOG_EXPORT_S3_KEY is required when " "ENABLE_DYNAMODB_USAGE_LOG_EXPORT=True." ) pi_routing = (values.get("PI_ALB_ROUTING") or "").strip().lower() if pi_ecs: if pi_routing not in PI_ALB_ROUTING_MODES: errors.append( f"PI_ALB_ROUTING must be one of {list(PI_ALB_ROUTING_MODES)}; got '{pi_routing}'." ) elif ( pi_routing in ("host", "both") and not (values.get("PI_ALB_HOST_HEADER") or "").strip() ): errors.append( "PI_ALB_HOST_HEADER is required when PI_ALB_ROUTING is 'host' or 'both'." ) return errors def build_app_config_env_values(values: Dict[str, str]) -> Dict[str, str]: """AWS deployment keys merged into config/app_config.env for ECS tasks.""" prefix_lower = (values.get("CDK_PREFIX") or "").lower() headless = values.get("ENABLE_HEADLESS_DEPLOYMENT") == "True" pi_express = values.get("ENABLE_PI_AGENT_EXPRESS_SERVICE") == "True" def _name(env_key: str, suffix: str) -> str: return (values.get(env_key) or "").strip() or f"{prefix_lower}{suffix}" output_bucket = _name("S3_OUTPUT_BUCKET_NAME", "s3-output") return { "COGNITO_AUTH": "False" if headless or pi_express else "True", "RUN_AWS_FUNCTIONS": "True", "RUN_AWS_BEDROCK_MODELS": "True", "RUN_LOCAL_MODEL": "False", "RUN_GEMINI_MODELS": "False", "RUN_AZURE_MODELS": "False", "PRIORITISE_SSO_OVER_AWS_ENV_ACCESS_KEYS": "True", "DISPLAY_FILE_NAMES_IN_LOGS": "False", "SESSION_OUTPUT_FOLDER": "True", "SAVE_LOGS_TO_CSV": "True", "SAVE_LOGS_TO_DYNAMODB": "True", "SHOW_COSTS": "True", "S3_LOG_BUCKET": _name("S3_LOG_CONFIG_BUCKET_NAME", "s3-logs"), "S3_OUTPUTS_BUCKET": output_bucket if headless else "", "S3_OUTPUTS_FOLDER": "output/" if headless else "", "SAVE_OUTPUTS_TO_S3": "True" if headless else "False", "ACCESS_LOG_DYNAMODB_TABLE_NAME": _name( "ACCESS_LOG_DYNAMODB_TABLE_NAME", "dynamodb-access-logs" ), "FEEDBACK_LOG_DYNAMODB_TABLE_NAME": _name( "FEEDBACK_LOG_DYNAMODB_TABLE_NAME", "dynamodb-feedback-logs" ), "USAGE_LOG_DYNAMODB_TABLE_NAME": _name( "USAGE_LOG_DYNAMODB_TABLE_NAME", "dynamodb-usage-logs" ), } def write_app_config_env_file( answers: InstallAnswers, values: Dict[str, str], *, overwrite: bool = False, ) -> Optional[Path]: if not answers.write_app_config_env: return None updates = build_app_config_env_values(values) APP_CONFIG_ENV_PATH.parent.mkdir(parents=True, exist_ok=True) if APP_CONFIG_ENV_PATH.is_file() and not overwrite: existing = read_env_file(APP_CONFIG_ENV_PATH) existing.update(updates) backup_file(APP_CONFIG_ENV_PATH) lines = [f"{k}={v}" for k, v in existing.items()] APP_CONFIG_ENV_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8") print(f"Updated {APP_CONFIG_ENV_PATH} (AWS deployment keys merged)") return APP_CONFIG_ENV_PATH if APP_CONFIG_ENV_EXAMPLE.is_file() and not overwrite: if APP_CONFIG_ENV_PATH.is_file(): backup_file(APP_CONFIG_ENV_PATH) shutil.copy2(APP_CONFIG_ENV_EXAMPLE, APP_CONFIG_ENV_PATH) existing = read_env_file(APP_CONFIG_ENV_PATH) existing.update(updates) lines = [f"{k}={v}" for k, v in existing.items()] APP_CONFIG_ENV_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8") print(f"Created {APP_CONFIG_ENV_PATH} from example + AWS defaults") return APP_CONFIG_ENV_PATH if APP_CONFIG_ENV_PATH.is_file(): backup_file(APP_CONFIG_ENV_PATH) lines = [ "# Generated by cdk_install.py — app runtime config for AWS ECS", *[f"{k}={v}" for k, v in updates.items()], ] APP_CONFIG_ENV_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8") print(f"Wrote {APP_CONFIG_ENV_PATH}") return APP_CONFIG_ENV_PATH def build_pi_agent_env_values(answers: InstallAnswers) -> Dict[str, str]: """Runtime settings for the Pi agent Gradio app (uploaded to S3 as pi_agent.env).""" values = { "PI_DEPLOYMENT_PROFILE": "aws-ecs", "PI_DEFAULT_PROVIDER": answers.pi_default_provider, "DOC_SUMMARISATION_GRADIO_URL": f"http://{answers.sc_discovery_name}:7860", "RUN_AWS_FUNCTIONS": "True", "AWS_REGION": answers.aws_region, "PI_GRADIO_PORT": answers.pi_gradio_port, "PI_DEFAULT_OCR_METHOD": "AWS Textract service - all PDF types", "PI_DEFAULT_PII_METHOD": "AWS Comprehend", } if answers.enable_pi_express: values["RUN_FASTAPI"] = "True" return values path_prefix = normalize_pi_path_prefix(answers.pi_alb_path_prefix) if answers.pi_alb_routing.strip().lower() in ("path", "both"): values["PI_ROOT_PATH"] = path_prefix values["ROOT_PATH"] = path_prefix values["FASTAPI_ROOT_PATH"] = path_prefix return values def write_pi_agent_env_file( answers: InstallAnswers, *, overwrite: bool = False, ) -> Optional[Path]: if not answers.pi_enabled or not answers.write_pi_agent_env: return None updates = build_pi_agent_env_values(answers) PI_AGENT_ENV_PATH.parent.mkdir(parents=True, exist_ok=True) if PI_AGENT_ENV_PATH.is_file() and not overwrite: existing = read_env_file(PI_AGENT_ENV_PATH) existing.update(updates) backup_file(PI_AGENT_ENV_PATH) lines = [f"{k}={v}" for k, v in existing.items()] PI_AGENT_ENV_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8") print(f"Updated {PI_AGENT_ENV_PATH} (AWS/Pi agent keys merged)") return PI_AGENT_ENV_PATH if PI_AGENT_ENV_EXAMPLE.is_file() and not overwrite: backup_file(PI_AGENT_ENV_PATH) if PI_AGENT_ENV_PATH.is_file() else None shutil.copy2(PI_AGENT_ENV_EXAMPLE, PI_AGENT_ENV_PATH) existing = read_env_file(PI_AGENT_ENV_PATH) existing.update(updates) lines = [f"{k}={v}" for k, v in existing.items()] PI_AGENT_ENV_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8") print(f"Created {PI_AGENT_ENV_PATH} from example + AWS defaults") return PI_AGENT_ENV_PATH backup_file(PI_AGENT_ENV_PATH) if PI_AGENT_ENV_PATH.is_file() else None lines = [ "# Generated by cdk_install.py — Pi agent runtime config for AWS ECS", *[f"{k}={v}" for k, v in updates.items()], ] PI_AGENT_ENV_PATH.write_text("\n".join(lines) + "\n", encoding="utf-8") print(f"Wrote {PI_AGENT_ENV_PATH}") return PI_AGENT_ENV_PATH def write_env_file(path: Path, values: Dict[str, str]) -> Path: backup_file(path) lines = [f"{key}={val}" for key, val in values.items()] path.parent.mkdir(parents=True, exist_ok=True) path.write_text("\n".join(lines) + "\n", encoding="utf-8") print(f"Wrote {path}") return path def read_env_file(path: Path) -> Dict[str, str]: values: Dict[str, str] = {} if not path.is_file(): return values for line in path.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line or line.startswith("#"): continue if "=" in line: key, _, val = line.partition("=") values[key.strip()] = val.strip() return values def patch_env_file(path: Path, updates: Dict[str, str]) -> None: values = read_env_file(path) values.update(updates) write_env_file(path, values) def print_summary(values: Dict[str, str], python_exe: Optional[Path] = None) -> None: keys = [ "CDK_PREFIX", "AWS_REGION", "AWS_ACCOUNT_ID", "USE_ECS_EXPRESS_MODE", "USE_CLOUDFRONT", "ENABLE_HEADLESS_DEPLOYMENT", "ENABLE_S3_BATCH_ECS_TRIGGER", "ENABLE_BEDROCK_MODEL_INVOCATION_LOGGING", "ENABLE_APPREGISTRY", "ENABLE_APPLICATION_RESOURCE_GROUP", "APPLICATION_NAME", "ENABLE_RESOURCE_DELETE_PROTECTION", "VPC_NAME", "NEW_VPC_CIDR", "ACM_SSL_CERTIFICATE_ARN", "SSL_CERTIFICATE_DOMAIN", "ENABLE_PI_AGENT_EXPRESS_SERVICE", "ENABLE_PI_AGENT_ECS_SERVICE", ] print("\n--- Configuration summary ---") for key in keys: if key in values: print(f" {key}={values[key]}") if python_exe: print(f" cdk.json app={format_cdk_app_command(python_exe)}") print("----------------------------\n") # --------------------------------------------------------------------------- # Deploy runner # --------------------------------------------------------------------------- def _deploy_env() -> Dict[str, str]: return build_cdk_subprocess_env() def run_cdk_command( args: List[str], *, check: bool = True ) -> subprocess.CompletedProcess: cdk_exe = resolve_cdk_executable() cmd = [cdk_exe, *args] print(f"Running: {' '.join(cmd)}") return subprocess.run( cmd, cwd=str(CDK_DIR), env=_deploy_env(), check=check, ) @dataclass(frozen=True) class ExistingStack: name: str region: str status: str termination_protection: bool = False def _should_check_cloudfront_stack( config_values: Optional[Dict[str, str]] = None, ) -> bool: """Skip us-east-1 CloudFront stack lookup when config disables that path.""" if not config_values: return True use_cloudfront = config_values.get("USE_CLOUDFRONT") run_useast = config_values.get("RUN_USEAST_STACK") if use_cloudfront is not None and use_cloudfront != "True": return False if run_useast is not None and run_useast != "True": return False return True def _stack_check_skippable_error(exc: Exception) -> bool: """Permission / SCP errors in one region must not block checks in others.""" from botocore.exceptions import ClientError if not isinstance(exc, ClientError): return False code = exc.response.get("Error", {}).get("Code", "") return code in ( "AccessDenied", "UnauthorizedOperation", "AuthorizationError", "AccessDeniedException", ) def derived_appregistry_stack_name( config_values: Optional[Dict[str, str]] = None, ) -> Optional[str]: """Resolve AppRegistry stack name from config (explicit or CDK_PREFIX default).""" if not config_values: return None explicit = (config_values.get("APPREGISTRY_STACK_NAME") or "").strip() if explicit: return explicit prefix = (config_values.get("CDK_PREFIX") or "").strip() if prefix: return f"{prefix}{APPREGISTRY_STACK_SUFFIX}" return None def list_regional_appregistry_stack_names(region: str) -> List[str]: """List active AppRegistry stacks in the regional account (suffix match).""" import boto3 from botocore.exceptions import ClientError cfn = boto3.client("cloudformation", region_name=region) skip_statuses = {"DELETE_COMPLETE", "DELETE_IN_PROGRESS"} names: List[str] = [] try: paginator = cfn.get_paginator("list_stacks") for page in paginator.paginate(): for summary in page.get("StackSummaries", []): name = summary.get("StackName", "") status = summary.get("StackStatus", "") if not name.endswith(APPREGISTRY_STACK_SUFFIX): continue if status in skip_statuses: continue names.append(name) except ClientError as exc: if _stack_check_skippable_error(exc): code = exc.response.get("Error", {}).get("Code", "ClientError") print( f"Warning: could not list AppRegistry stacks in {region} " f"({code}). Continuing with configured stack names only." ) return [] raise return sorted(dict.fromkeys(names)) def appregistry_stack_names_to_check( config_values: Optional[Dict[str, str]] = None, *, discovered_names: Optional[Sequence[str]] = None, ) -> List[str]: """Merge configured and discovered AppRegistry stack names (deduped).""" names: List[str] = [] derived = derived_appregistry_stack_name(config_values) if derived: names.append(derived) for name in discovered_names or (): if name not in names: names.append(name) return names def stacks_to_check( regional_region: str, config_values: Optional[Dict[str, str]] = None, *, discovered_appregistry_names: Optional[Sequence[str]] = None, ) -> List[Tuple[str, str]]: """Return (stack_name, region) pairs in safe deletion order.""" checks: List[Tuple[str, str]] = [] if _should_check_cloudfront_stack(config_values): checks.append((CLOUDFRONT_STACK, CLOUDFRONT_STACK_REGION)) for appregistry_name in appregistry_stack_names_to_check( config_values, discovered_names=discovered_appregistry_names, ): checks.append((appregistry_name, regional_region)) checks.append((REGIONAL_STACK, regional_region)) return checks def describe_existing_stack(stack_name: str, region: str) -> Optional[ExistingStack]: import boto3 from botocore.exceptions import ClientError cfn = boto3.client("cloudformation", region_name=region) try: response = cfn.describe_stacks(StackName=stack_name) except ClientError as exc: code = exc.response.get("Error", {}).get("Code", "") if code in ("ValidationError", "ResourceNotFoundException") or ( "does not exist" in str(exc) ): return None raise stack = response["Stacks"][0] status = stack.get("StackStatus", "") if status == "DELETE_COMPLETE": return None return ExistingStack( name=stack_name, region=region, status=status, termination_protection=bool(stack.get("EnableTerminationProtection", False)), ) def discover_existing_llm_topic_stacks( regional_region: str, config_values: Optional[Dict[str, str]] = None, ) -> List[ExistingStack]: """Find deployed llm_topic_modeller CloudFormation stacks in the target account.""" from botocore.exceptions import ClientError discovered_appregistry = list_regional_appregistry_stack_names(regional_region) checks = stacks_to_check( regional_region, config_values, discovered_appregistry_names=discovered_appregistry, ) ordered: List[ExistingStack] = [] for stack_name, region in checks: try: info = describe_existing_stack(stack_name, region) except ClientError as exc: if _stack_check_skippable_error(exc): code = exc.response.get("Error", {}).get("Code", "ClientError") print( f"Warning: could not check stack {stack_name!r} in {region} " f"({code}). Continuing with other regions." ) continue raise if info: ordered.append(info) return ordered def discover_existing_doc_summarisation_stacks( regional_region: str, config_values: Optional[Dict[str, str]] = None, ) -> List[ExistingStack]: """Deprecated alias for discover_existing_llm_topic_stacks.""" return discover_existing_llm_topic_stacks(regional_region, config_values) def force_delete_cloudformation_stacks( stacks: Sequence[ExistingStack], *, wait: bool = True, ) -> None: """Delete stacks via CloudFormation (disables termination protection first).""" import boto3 from botocore.exceptions import ClientError clients: Dict[str, Any] = {} def cfn_for(region: str): if region not in clients: clients[region] = boto3.client("cloudformation", region_name=region) return clients[region] for stack in stacks: cfn = cfn_for(stack.region) if stack.termination_protection: print(f"Disabling termination protection on {stack.name} ({stack.region})") cfn.update_termination_protection( EnableTerminationProtection=False, StackName=stack.name, ) if stack.status == "DELETE_IN_PROGRESS": print(f"{stack.name} ({stack.region}) is already deleting.") else: print(f"Deleting stack {stack.name} ({stack.region}) ...") try: cfn.delete_stack(StackName=stack.name) except ClientError as exc: code = exc.response.get("Error", {}).get("Code", "") if code == "ValidationError" and "DELETE_IN_PROGRESS" in str(exc): print(f"{stack.name} is already deleting.") else: raise if not wait: return for stack in stacks: cfn = cfn_for(stack.region) print(f"Waiting for {stack.name} ({stack.region}) to finish deleting ...") try: cfn.get_waiter("stack_delete_complete").wait( StackName=stack.name, WaiterConfig={"Delay": 15, "MaxAttempts": 120}, ) except ClientError: detail = describe_existing_stack(stack.name, stack.region) status = detail.status if detail else "unknown" raise SystemExit( f"Stack {stack.name} ({stack.region}) did not delete cleanly " f"(status={status}). Resolve in the CloudFormation console and retry." ) from None def handle_existing_stacks_at_start( args: argparse.Namespace, regional_region: str, *, config_values: Optional[Dict[str, str]] = None, ) -> None: """At wizard start: report existing stacks and optionally force-delete them.""" if getattr(args, "skip_stack_check", False) or args.config_only or args.synth_only: return if config_values is None and ENV_PATH.is_file(): config_values = read_env_file(ENV_PATH) try: existing = discover_existing_llm_topic_stacks(regional_region, config_values) except Exception as exc: print(f"Existing stack check skipped: {exc}") return if not existing: return print("\n--- Existing llm_topic_modeller CloudFormation stacks ---") for stack in existing: line = f" {stack.name} ({stack.region}): {stack.status}" if stack.termination_protection: line += " [termination protection ON]" print(line) if args.force_delete_stacks: should_delete = True elif args.yes: print( "\nStacks already exist in this account/region. " "Pass --force-delete-stacks to remove them before deploy, " "or omit it to update in place." ) return else: should_delete = ask_yes_no( "Force-delete these stacks before continuing? " "(disables termination protection and deletes all stack resources)", default=False, ) if not should_delete: print("Keeping existing stacks (deploy will update them in place).\n") return if not args.force_delete_stacks and not args.yes: if not ask_yes_no( "This permanently deletes AWS resources in these stacks. Proceed?", default=False, ): print("Stack deletion cancelled.\n") return force_delete_cloudformation_stacks(existing) print("Existing stacks deleted.\n") def fetch_stack_output( stack_name: str, output_key: str, region: str, ) -> Optional[str]: import boto3 from botocore.exceptions import ClientError cfn = boto3.client("cloudformation", region_name=region) try: response = cfn.describe_stacks(StackName=stack_name) except ClientError: return None for stack in response.get("Stacks", []): for output in stack.get("Outputs", []): if output.get("OutputKey") == output_key: return output.get("OutputValue") return None def apply_post_deploy_fixup(values: Dict[str, str], assume_yes: bool) -> bool: """Return True if post-deploy fixup changed env and/or Cognito callbacks.""" from cdk_post_deploy import ( apply_cognito_alb_callback_fixup, cognito_alb_callbacks_need_update, ) if values.get("ENABLE_HEADLESS_DEPLOYMENT") == "True": return False region = values.get("AWS_REGION", "") express = values.get("USE_ECS_EXPRESS_MODE") == "True" cloudfront = values.get("USE_CLOUDFRONT") == "True" fixup_applied = False pool_id = fetch_stack_output(REGIONAL_STACK, "CognitoPoolId", region) client_id = fetch_stack_output(REGIONAL_STACK, "CognitoAppClientId", region) if not pool_id or not client_id: print( "CognitoPoolId or CognitoAppClientId stack output missing; " "skipping Cognito callback fixup." ) return False if express and not cloudfront: from cdk_config import normalize_https_redirect_url endpoint = fetch_stack_output(REGIONAL_STACK, "ExpressServiceEndpoint", region) if not endpoint: print("No ExpressServiceEndpoint output found; skipping Cognito URL fixup.") return False endpoint = normalize_https_redirect_url(endpoint) current = (values.get("ECS_EXPRESS_COGNITO_REDIRECT_BASE") or "").strip() env_needs_patch = current != endpoint cognito_needs_update = cognito_alb_callbacks_need_update( pool_id, client_id, endpoint, aws_region=region ) if not env_needs_patch and not cognito_needs_update: print("Express Cognito redirect base and callback URLs already set.") else: if env_needs_patch: print(f"Setting ECS_EXPRESS_COGNITO_REDIRECT_BASE={endpoint}") patch_env_file( ENV_PATH, { "ECS_EXPRESS_COGNITO_REDIRECT_BASE": endpoint, "COGNITO_REDIRECTION_URL": endpoint, }, ) fixup_applied = True if cognito_needs_update and ( assume_yes or ask_yes_no( "Update Cognito app client callback URLs for the Express endpoint " "(no CDK redeploy)?", True, ) ): if apply_cognito_alb_callback_fixup( user_pool_id=pool_id, client_id=client_id, redirect_base=endpoint, aws_region=region, ): fixup_applied = True elif env_needs_patch and cognito_needs_update: print( "Env updated with Express URL; Cognito callbacks unchanged. " "ALB login will fail until callback URLs are updated." ) elif cloudfront: cf_domain = fetch_stack_output( CLOUDFRONT_STACK, "CloudFrontDistributionURL", CLOUDFRONT_STACK_REGION ) if not cf_domain: print( "No CloudFrontDistributionURL output found; skipping CloudFront fixup." ) return fixup_applied redirect_base = f"https://{cf_domain.strip()}" current = (values.get("CLOUDFRONT_DOMAIN") or "").strip() env_needs_patch = current != cf_domain.strip() cognito_needs_update = cognito_alb_callbacks_need_update( pool_id, client_id, redirect_base, aws_region=region ) if not env_needs_patch and not cognito_needs_update: print("CloudFront domain and Cognito callback URLs already set.") return fixup_applied if env_needs_patch: print(f"Setting CLOUDFRONT_DOMAIN={cf_domain}") patch_env_file( ENV_PATH, { "CLOUDFRONT_DOMAIN": cf_domain, "COGNITO_REDIRECTION_URL": redirect_base, }, ) fixup_applied = True if cognito_needs_update and ( assume_yes or ask_yes_no( "Update Cognito app client callback URLs for the CloudFront domain " "(no CDK redeploy)?", True, ) ): if apply_cognito_alb_callback_fixup( user_pool_id=pool_id, client_id=client_id, redirect_base=redirect_base, aws_region=region, ): fixup_applied = True elif env_needs_patch and cognito_needs_update: print( "Env updated with CloudFront domain; Cognito callbacks unchanged. " "ALB login will fail until callback URLs are updated." ) if express: from cdk_post_deploy import apply_cognito_secret_fixup_from_stack resolved = resolve_fixup_env_values(values) cluster_name = resolved["CLUSTER_NAME"] main_service = resolved["ECS_EXPRESS_SERVICE_NAME"] secret_name = resolved.get("COGNITO_USER_POOL_CLIENT_SECRET_NAME", "") if not main_service: print( "ECS express service name could not be resolved; " "skipping Express Cognito secret sync." ) return fixup_applied try: if pool_id and client_id: if apply_cognito_secret_fixup_from_stack( stack_name=REGIONAL_STACK, secret_name=secret_name, cluster_name=cluster_name, main_service_name=main_service, aws_region=region, recycle_tasks=False, ): fixup_applied = True except Exception as exc: print(f"Warning: could not sync Express Cognito secret: {exc}") return fixup_applied def run_quickstart(python_exe: Path) -> None: if not QUICKSTART_SCRIPT.is_file(): raise SystemExit(f"Quickstart script not found: {QUICKSTART_SCRIPT}") cmd = [str(python_exe), str(QUICKSTART_SCRIPT)] print(f"Running: {' '.join(cmd)}") subprocess.run(cmd, cwd=str(CDK_DIR), env=_deploy_env(), check=True) # --------------------------------------------------------------------------- # Pi agent configuration (disabled in installer; stack code retained for future) # --------------------------------------------------------------------------- _DEPRECATED_PI_CLI_FLAGS = ( "--enable-pi", "--enable-pi-express", "--enable-pi-legacy", "--pi-alb-routing", "--pi-path-prefix", "--pi-host-header", "--pi-listener-priority", "--pi-gradio-port", "--sc-discovery-name", "--pi-provider", "--skip-pi-agent-env", ) def warn_deprecated_pi_cli_flags(args: argparse.Namespace) -> None: """Ignore Pi agent CLI flags — not supported in llm_topic_modeller yet.""" argv = getattr(args, "_argv", ()) or () if any( token == flag or token.startswith(f"{flag}=") for flag in _DEPRECATED_PI_CLI_FLAGS for token in argv ): print( "Note: Pi agent is not supported in llm_topic_modeller yet; " "ignoring Pi-related flags." ) answers_pi = ( getattr(args, "enable_pi", False) or getattr(args, "enable_pi_express", False) or getattr(args, "enable_pi_legacy", False) ) if answers_pi: print( "Note: Pi agent is not supported in llm_topic_modeller yet; " "ignoring Pi-related flags." ) def apply_pi_cli_flags(args: argparse.Namespace, answers: InstallAnswers) -> None: if getattr(args, "enable_pi", False): preset = merge_preset(answers.profile, answers.custom_overrides) if preset.get("USE_ECS_EXPRESS_MODE") == "True": answers.enable_pi_express = True else: answers.enable_pi_legacy = True answers.enable_service_connect = True if getattr(args, "enable_pi_express", False): answers.enable_pi_express = True if getattr(args, "enable_pi_legacy", False): answers.enable_pi_legacy = True answers.enable_service_connect = True if args.pi_alb_routing: answers.pi_alb_routing = args.pi_alb_routing if args.pi_path_prefix: answers.pi_alb_path_prefix = args.pi_path_prefix if args.pi_host_header: answers.pi_alb_host_header = args.pi_host_header if args.pi_listener_priority: answers.pi_alb_listener_rule_priority = args.pi_listener_priority if args.pi_gradio_port: answers.pi_gradio_port = args.pi_gradio_port if args.sc_discovery_name: answers.sc_discovery_name = args.sc_discovery_name if args.pi_provider: answers.pi_default_provider = args.pi_provider if args.skip_pi_agent_env: answers.write_pi_agent_env = False def configure_app_config_options( answers: InstallAnswers, args: argparse.Namespace, *, interactive: bool, ) -> None: if getattr(args, "skip_app_config_env", False): answers.write_app_config_env = False return if getattr(args, "overwrite_app_config_env", False): answers.write_app_config_env = True answers.overwrite_app_config_env = True return if not interactive: answers.write_app_config_env = True return example_label = ( APP_CONFIG_ENV_EXAMPLE.name if APP_CONFIG_ENV_EXAMPLE.is_file() else "defaults" ) if ask_yes_no( f"Write/update {APP_CONFIG_ENV_PATH.name} for the deployed app " f"(from {example_label} + AWS resource names)?", default=True, ): answers.write_app_config_env = True if APP_CONFIG_ENV_PATH.is_file(): answers.overwrite_app_config_env = ask_yes_no( f"{APP_CONFIG_ENV_PATH.name} exists — replace file from example template?", default=False, ) else: answers.write_app_config_env = False def configure_pi_options( answers: InstallAnswers, args: argparse.Namespace, *, interactive: bool, assume_yes: bool, ) -> None: preset = merge_preset(answers.profile, answers.custom_overrides) use_express = preset.get("USE_ECS_EXPRESS_MODE") == "True" use_cloudfront = preset.get("USE_CLOUDFRONT") == "True" apply_pi_cli_flags(args, answers) if not answers.pi_enabled and interactive: if use_express: label = ( "Deploy agent mode (second Gradio app on Express, dedicated HTTPS URL)?" ) answers.enable_pi_express = ask_yes_no(label, default=False) else: label = "Deploy agent mode (second Gradio app on legacy Fargate + Service Connect)?" if ask_yes_no(label, default=False): answers.enable_pi_legacy = True answers.enable_service_connect = True if not answers.pi_enabled: return if not answers.pi_alb_listener_rule_priority and not answers.enable_pi_express: answers.pi_alb_listener_rule_priority = default_pi_listener_priority( use_cloudfront ) if answers.enable_pi_express: if interactive and not assume_yes: print("\n--- Agent mode (ECS Express) ---") answers.sc_discovery_name = ask( "Service Connect discovery name for main app", answers.sc_discovery_name, ) answers.pi_gradio_port = ask("Agent Gradio port", answers.pi_gradio_port) if ask_yes_no( f"Write/update {PI_AGENT_ENV_PATH.name} for AWS ECS (DOC_SUMMARISATION_GRADIO_URL, etc.)?", default=True, ): answers.write_pi_agent_env = True if PI_AGENT_ENV_PATH.is_file(): answers.overwrite_pi_agent_env = ask_yes_no( "pi_agent.env exists — replace file from example template?", default=False, ) else: answers.write_pi_agent_env = False print( f"Agent mode: Express (dedicated HTTPS endpoint per service); " f"Service Connect discovery={answers.sc_discovery_name}" ) return if interactive and not ( args.pi_alb_routing or args.pi_path_prefix or args.pi_host_header or assume_yes ): print("\n--- Agent mode ALB routing ---") ridx = ask_choice( "How should the shared ALB route traffic to the Agent UI?", [ "Path prefix (default /agent/ — e.g. https://host/agent/)", "Dedicated hostname (PI_ALB_HOST_HEADER)", "Both path prefix and hostname", ], default_index=0, ) answers.pi_alb_routing = PI_ALB_ROUTING_MODES[ridx] if answers.pi_alb_routing in ("path", "both"): answers.pi_alb_path_prefix = ask( "Agent path prefix", answers.pi_alb_path_prefix, ) if answers.pi_alb_routing in ("host", "both"): default_host = "" if answers.ssl_domain: default_host = f"agent.{answers.ssl_domain}" answers.pi_alb_host_header = ask( "Agent ALB host header (DNS CNAME to CloudFront/ALB)", default_host, ) if use_cloudfront: default_pri = default_pi_listener_priority(True) answers.pi_alb_listener_rule_priority = ask( "Agent ALB listener rule priority (default 3; priorities 1–2 reserved)", default_pri, ) answers.sc_discovery_name = ask( "Service Connect discovery name for main app", answers.sc_discovery_name, ) answers.pi_gradio_port = ask("Agent Gradio port", answers.pi_gradio_port) if ask_yes_no( f"Write/update {PI_AGENT_ENV_PATH.name} for AWS ECS (DOC_SUMMARISATION_GRADIO_URL, etc.)?", default=True, ): answers.write_pi_agent_env = True if PI_AGENT_ENV_PATH.is_file(): answers.overwrite_pi_agent_env = ask_yes_no( "pi_agent.env exists — replace file from example template?", default=False, ) else: answers.write_pi_agent_env = False print( f"Agent mode: " f"{'Express' if answers.enable_pi_express else 'legacy Fargate'}, " f"routing={answers.pi_alb_routing}, " f"prefix={normalize_pi_path_prefix(answers.pi_alb_path_prefix)}" ) # --------------------------------------------------------------------------- # Interactive wizard # --------------------------------------------------------------------------- def run_wizard(args: argparse.Namespace) -> InstallAnswers: answers = InstallAnswers() interactive = not args.yes assume_yes = args.yes if args.profile: answers.profile = args.profile elif interactive: idx = ask_choice( "Deployment profile", [ "Demonstration (Express, no CloudFront, no delete protection)", "Production (ACM cert, CloudFront, delete protection)", "Headless batch only (demo-style networking, no Express web UI)", "Custom (configure individual toggles)", ], ) answers.profile = ("demo", "production", "headless", "custom")[idx] else: answers.profile = "demo" if answers.profile == "headless": answers.enable_headless = True elif getattr(args, "headless", False): if answers.profile == "demo": raise SystemExit( "Headless batch mode is not available with --profile demo (Express). " "Use --profile headless, --profile production --headless, " "or --profile custom without ECS Express." ) answers.enable_headless = True elif interactive and answers.profile == "production": answers.enable_headless = ask_yes_no( "Enable headless batch-only deployment (S3 → Lambda → one-shot ECS, " "no always-on web UI)?", default=False, ) if answers.profile == "custom" and interactive: answers.custom_overrides["USE_ECS_EXPRESS_MODE"] = ( "True" if ask_yes_no("Use ECS Express Mode?", False) else "False" ) answers.custom_overrides["USE_CLOUDFRONT"] = ( "True" if ask_yes_no("Use CloudFront?", True) else "False" ) answers.custom_overrides["RUN_USEAST_STACK"] = ( "True" if answers.custom_overrides.get("USE_CLOUDFRONT") == "True" and ask_yes_no( "Deploy us-east-1 CloudFront stack (RUN_USEAST_STACK)?", True ) else "False" ) answers.custom_overrides["ENABLE_RESOURCE_DELETE_PROTECTION"] = ( "True" if ask_yes_no("Enable delete protection?", True) else "False" ) if getattr(args, "headless", False): if answers.custom_overrides.get("USE_ECS_EXPRESS_MODE") == "True": raise SystemExit( "Headless batch mode requires USE_ECS_EXPRESS_MODE=False in " "the custom profile." ) answers.enable_headless = True elif profile_allows_headless_add_on(answers): answers.enable_headless = ask_yes_no( "Enable headless batch-only deployment (S3 → Lambda → one-shot ECS, " "no always-on web UI)?", default=False, ) headless_err = headless_profile_error(answers) if headless_err: raise SystemExit(headless_err) try: account, region = get_aws_identity(args.region) except Exception as exc: raise SystemExit(f"AWS credentials error: {exc}") from exc answers.aws_account_id = args.account or account answers.aws_region = args.region or region if interactive and not args.account: answers.aws_account_id = ask("AWS account ID", answers.aws_account_id) if interactive and not args.region: answers.aws_region = ask("AWS region", answers.aws_region) default_prefix = f"{answers.profile.capitalize()}-Summarisation-" answers.cdk_prefix = args.cdk_prefix or ( ask("CDK resource prefix (e.g. MyOrg-Summarisation-)", default_prefix) if interactive else default_prefix ) if not answers.cdk_prefix.endswith("-"): answers.cdk_prefix += "-" if answers_use_headless(answers): answers.cognito_domain_prefix = "" print("Headless batch mode: skipping Cognito hosted UI domain (no web login).") resolve_globally_unique_install_names( answers, interactive=interactive, assume_yes=assume_yes, args=args, ) # VPC if args.new_vpc_cidr: answers.vpc_mode = "new" answers.new_vpc_cidr = args.new_vpc_cidr elif args.vpc_name: answers.vpc_mode = "existing" answers.vpc_name = args.vpc_name elif interactive: vpc_idx = ask_choice( "VPC", ["Create new VPC", "Use existing VPC"], ) answers.vpc_mode = "new" if vpc_idx == 0 else "existing" else: answers.vpc_mode = "existing" if answers.vpc_mode == "new": prompt_new_vpc_cidr(answers, interactive=interactive) else: if not answers.vpc_name: vpcs = list_vpcs(answers.aws_region) if not vpcs: raise SystemExit("No VPCs found in region.") if interactive: labels = [f"{v['name']} ({v['cidr']})" for v in vpcs] vidx = ask_choice("Select VPC", labels) answers.vpc_name = vpcs[vidx]["name"] else: raise SystemExit( "--vpc-name required for existing VPC in non-interactive mode." ) vpc_list = list_vpcs(answers.aws_region) vpc_id = next( (v["id"] for v in vpc_list if v["name"] == answers.vpc_name), None, ) # Subnets (public only for Express or demo-style headless) apply_subnet_cli_flags(args, answers) public_subnets_only = answers_use_public_subnets_only(answers) if public_subnets_only: if interactive: answers.public_subnet_mode = ask_subnet_tier_mode("Public") else: answers.public_subnet_mode, _ = resolve_subnet_tier_modes(args) answers.private_subnet_mode = "auto" answers.private_subnet_names = [] answers.private_subnet_cidrs = [] answers.private_subnet_azs = [] elif interactive: layout_idx = ask_choice( "Subnets in existing VPC", [ "Auto-discover public and private subnets", "Use existing named subnets (both tiers)", "Create new stack-specific subnets (both tiers)", "Configure public and private separately", ], default_index=0, ) if layout_idx < 3: shared_mode = SUBNET_TIER_MODES[layout_idx] answers.public_subnet_mode = shared_mode answers.private_subnet_mode = shared_mode else: answers.public_subnet_mode = ask_subnet_tier_mode("Public") answers.private_subnet_mode = ask_subnet_tier_mode("Private") else: answers.public_subnet_mode, answers.private_subnet_mode = ( resolve_subnet_tier_modes(args) ) azs = list_availability_zones(answers.aws_region)[:3] vpc_cidr = next( (v["cidr"] for v in vpc_list if v["name"] == answers.vpc_name), "", ) if vpc_id: subnets = list_subnets_in_vpc(vpc_id, answers.aws_region) configure_subnet_tier( answers, "public", answers.public_subnet_mode, subnets, azs, interactive=interactive, vpc_cidr=vpc_cidr, ) if not public_subnets_only: reserved_public = ( list(answers.public_subnet_cidrs) if answers.public_subnet_mode == "create" else [] ) configure_subnet_tier( answers, "private", answers.private_subnet_mode, subnets, azs, interactive=interactive, vpc_cidr=vpc_cidr, reserved_subnet_cidrs=reserved_public, ) # Optional infra reuse if ( vpc_id and interactive and ask_yes_no("Configure optional existing infra (IGW/ALB)?", False) ): igws = list_igws_for_vpc(vpc_id, answers.aws_region) if igws: # igws_ids = [g["id"] for g in igws] print("Internet gateways:", ", ".join(g["id"] for g in igws)) answers.existing_igw_id = ask("EXISTING_IGW_ID (blank=new)", "") albs = list_albs_in_vpc(vpc_id, answers.aws_region) if albs: labels = [f"{a['name']} ({a['dns']})" for a in albs] if ask_yes_no("Reuse an existing ALB?", False): aidx = ask_choice("Select ALB", labels) answers.existing_alb_arn = albs[aidx]["arn"] answers.existing_alb_dns = albs[aidx]["dns"] # Production TLS preset = merge_preset(answers_preset_profile(answers), answers.custom_overrides) use_express = preset.get("USE_ECS_EXPRESS_MODE") == "True" if ( preset.get("USE_CLOUDFRONT") == "True" and preset.get("USE_ECS_EXPRESS_MODE") != "True" and not answers_use_headless(answers) ): if args.cert_arn: answers.acm_cert_arn = args.cert_arn answers.ssl_domain = args.domain or "" elif interactive: certs = list_acm_certificates(answers.aws_region) if not certs: raise SystemExit(f"No ISSUED ACM certificates in {answers.aws_region}.") labels = [f"{c['label']} — {c['arn']}" for c in certs] cidx = ask_choice("Select ACM certificate", labels) answers.acm_cert_arn = certs[cidx]["arn"] answers.ssl_domain = ask( "SSL certificate domain (Cognito/CloudFront URL)", certs[cidx]["domain"], ) else: raise SystemExit( "--cert-arn and --domain required for production in non-interactive mode." ) if interactive: answers.cloudfront_geo = ask( "CloudFront geo restriction (e.g. GB, blank=none)", "" ) if not answers_use_headless(answers): warn_deprecated_pi_cli_flags(args) # Advanced add-ons (non-Pi agent) is_express = use_express and not answers_use_headless(answers) if answers_use_headless(answers) and interactive: mem = ask("ECS task memory (MB)", answers.ecs_memory) if mem: answers.ecs_memory = mem if getattr(args, "headless_output_notifications", False): answers.enable_headless_output_notifications = True elif getattr(args, "no_headless_output_notifications", False): answers.enable_headless_output_notifications = False elif ask_yes_no( "Enable email notifications when new analysis outputs are uploaded to S3?", default=True, ): answers.enable_headless_output_notifications = True if answers.enable_headless_output_notifications: default_email = getattr(args, "headless_notify_email", "") or "" while True: answers.headless_output_notify_email = ask( "Notification email address (SNS subscription; confirm via AWS email)", default_email, ).strip() email_error = validate_notify_email( answers.headless_output_notify_email ) if not email_error: break print(email_error) default_iam_user = ( getattr(args, "headless_output_iam_user", "").strip() or f"{answers.cdk_prefix}s3-output-reader" ) iam_user = ask( "IAM user name for programmatic S3 output access", default_iam_user, ).strip() answers.headless_output_iam_user_name = iam_user or default_iam_user elif answers_use_headless(answers): if getattr(args, "headless_output_notifications", False): answers.enable_headless_output_notifications = True answers.headless_output_notify_email = ( getattr(args, "headless_notify_email", "") or "" ).strip() answers.headless_output_iam_user_name = ( getattr(args, "headless_output_iam_user", "").strip() or f"{answers.cdk_prefix}s3-output-reader" ) email_error = validate_notify_email(answers.headless_output_notify_email) if email_error: raise SystemExit(email_error) elif interactive and ask_yes_no("Configure other optional add-ons?", False): if not is_express: if not answers.enable_service_connect: answers.enable_service_connect = ask_yes_no( "Enable ECS Service Connect (without agent mode)?", False ) answers.enable_s3_batch = ask_yes_no("Enable S3 batch ECS trigger?", False) mem = ask("ECS task memory (MB)", answers.ecs_memory) if mem: answers.ecs_memory = mem prompt_dynamo_usage_log_export_options(answers, args, interactive=interactive) prompt_bedrock_model_invocation_logging_options( answers, args, interactive=interactive ) prompt_application_tagging_and_appregistry_options( answers, args, interactive=interactive ) configure_app_config_options(answers, args, interactive=interactive) answers.python_path = args.python return answers # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def build_arg_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( description="Interactive CDK installer for llm_topic_modeller", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) p.add_argument( "--profile", choices=("demo", "production", "headless", "custom"), help="Base profile; use --profile headless or production --headless for batch-only", ) p.add_argument( "--yes", action="store_true", help="Accept defaults / skip confirmations" ) p.add_argument("--config-only", action="store_true", help="Write config files only") p.add_argument( "--deploy-only", action="store_true", help="Skip wizard; use existing cdk_config.env", ) p.add_argument( "--synth-only", action="store_true", help="Run cdk synth only (no deploy)" ) p.add_argument("--skip-deploy", action="store_true", help="Skip cdk deploy") p.add_argument( "--skip-quickstart", action="store_true", help="Skip post_cdk_build_quickstart.py", ) p.add_argument( "--skip-stack-check", action="store_true", help="Do not check for existing CloudFormation stacks at startup", ) p.add_argument( "--force-delete-stacks", action="store_true", help="If llm_topic_modeller stacks already exist, delete them before continuing " "(disables termination protection; implies consent in non-interactive mode)", ) p.add_argument( "--skip-cdk-json", action="store_true", help="Do not update cdk.json" ) p.add_argument( "--refresh-cdk-json", action="store_true", help="Force rewrite cdk.json app key" ) p.add_argument("--python", help="Python executable for cdk.json") p.add_argument("--region", help="AWS region") p.add_argument("--account", help="AWS account ID") p.add_argument("--cdk-prefix", help="CDK resource prefix") p.add_argument("--cognito-prefix", help="Cognito user pool domain prefix") p.add_argument( "--s3-log-bucket", help="S3 log/config bucket name (globally unique; skips wizard suggestion)", ) p.add_argument( "--s3-output-bucket", help="S3 output bucket name (globally unique; skips wizard suggestion)", ) p.add_argument("--vpc-name", help="Existing VPC Name tag") p.add_argument("--new-vpc-cidr", help="CIDR for new VPC") p.add_argument( "--subnet-mode", choices=SUBNET_TIER_MODES, help="Subnet mode for both public and private tiers (overridden by per-tier flags)", ) p.add_argument( "--public-subnet-mode", choices=SUBNET_TIER_MODES, help="Public subnet tier: auto, existing, or create", ) p.add_argument( "--private-subnet-mode", choices=SUBNET_TIER_MODES, help="Private subnet tier: auto, existing, or create", ) p.add_argument( "--public-subnet-names", help="Comma-separated existing public subnet names (with --public-subnet-mode existing)", ) p.add_argument( "--private-subnet-names", help="Comma-separated existing private subnet names (with --private-subnet-mode existing)", ) p.add_argument( "--headless", action="store_true", help="Batch-only add-on (production/custom without Express only; not valid with demo)", ) p.add_argument( "--headless-output-notifications", action="store_true", help="Enable S3 output PutRequests alarm, SNS email, and IAM reader user (headless)", ) p.add_argument( "--no-headless-output-notifications", action="store_true", help="Disable headless output notifications in interactive headless installs", ) p.add_argument( "--headless-notify-email", help="Email for SNS notifications when headless outputs are uploaded", ) p.add_argument( "--headless-output-iam-user", help="IAM user name for programmatic download of headless S3 outputs", ) p.add_argument( "--dynamo-usage-log-export", action="store_true", help="Enable scheduled DynamoDB usage log export Lambda (EventBridge)", ) p.add_argument( "--no-dynamo-usage-log-export", action="store_true", help="Disable DynamoDB usage log export in interactive installs", ) p.add_argument( "--dynamo-export-schedule-time", metavar="HH:MM", help="UTC schedule time for DynamoDB usage log export (default: 06:00)", ) p.add_argument( "--dynamo-export-schedule-days", choices=tuple(DYNAMO_EXPORT_SCHEDULE_DAY_PRESETS), help="Days to run DynamoDB usage log export (default: daily)", ) p.add_argument( "--dynamo-export-s3-key", help="S3 object key for the exported CSV (default: reports/dynamodb-usage/...)", ) p.add_argument( "--dynamo-export-date-attribute", help="DynamoDB attribute used for date filtering (default: timestamp)", ) p.add_argument( "--bedrock-model-invocation-logging", action="store_true", help=( "Enable Bedrock model invocation logging to the S3 log/config bucket " "and CloudWatch Logs (account/Region-level)" ), ) p.add_argument( "--no-bedrock-model-invocation-logging", action="store_true", help="Disable Bedrock model invocation logging in interactive installs", ) p.add_argument( "--bedrock-invocation-s3-prefix", default="", help="S3 key prefix for Bedrock invocation logs (default: bedrock-logs)", ) p.add_argument( "--bedrock-invocation-log-group", default="", help="CloudWatch log group for Bedrock invocation logs", ) p.add_argument( "--bedrock-invocation-log-retention-days", type=int, default=None, metavar="DAYS", help="CloudWatch retention for Bedrock invocation logs (default: 90)", ) p.add_argument( "--appregistry", action="store_true", help=( "Enable legacy AppRegistry / myApplications stack " "(opt-in; new customers cut off after 30 Jul 2026)" ), ) p.add_argument( "--no-appregistry", action="store_true", help="Disable AppRegistry (default)", ) p.add_argument( "--application-resource-group", action="store_true", help=( "Tag CDK-managed resources and create a tag-based AWS Resource Group " "(default on)" ), ) p.add_argument( "--no-application-resource-group", action="store_true", help="Disable application Resource Group creation", ) p.add_argument( "--application-name", default="", help="Application tag value / Resource Group filter (default: {CDK_PREFIX}llm-topic-modeller)", ) p.add_argument( "--application-tag-key", default="", help="Application tag key applied to CDK-managed resources (default: Application)", ) p.add_argument( "--create-headless-output-access-key", action="store_true", help="After deploy, create IAM access key for the headless output reader user", ) p.add_argument( "--skip-headless-output-access-key", action="store_true", help="Do not offer to create IAM access key after headless deploy", ) p.add_argument("--cert-arn", help="ACM certificate ARN (production)") p.add_argument("--domain", help="SSL certificate domain (production)") p.add_argument( "--skip-app-config-env", action="store_true", help="Do not write cdk/config/app_config.env from app_config.env.example", ) p.add_argument( "--overwrite-app-config-env", action="store_true", help="Replace existing config/app_config.env from app_config.env.example (non-interactive)", ) return p def main(argv: Optional[Sequence[str]] = None) -> int: argv = list(argv) if argv is not None else None args = build_arg_parser().parse_args(argv) args._argv = tuple(argv if argv is not None else sys.argv[1:]) CONFIG_DIR.mkdir(parents=True, exist_ok=True) print("llm_topic_modeller CDK installer\n") # Prerequisites try: cdk_version = check_cdk_cli() print(f"CDK CLI: {cdk_version}") except SystemExit: if not args.config_only: raise print("Warning: CDK CLI not found; config-only mode continuing.") if not args.config_only and not args.synth_only: try: _stack_account, stack_region = get_aws_identity(args.region) handle_existing_stacks_at_start(args, stack_region) except SystemExit: raise except Exception as exc: print(f"Existing stack check skipped: {exc}") python_exe: Optional[Path] = None if not args.skip_cdk_json: python_exe = resolve_python_executable( override=args.python, interactive=not args.yes and not args.deploy_only, assume_yes=args.yes or args.deploy_only, ) write_cdk_json( python_exe, force=args.refresh_cdk_json or not CDK_JSON_PATH.is_file(), skip=args.skip_cdk_json, ) elif CDK_JSON_PATH.is_file(): # Best-effort parse python from existing cdk.json try: data = json.loads(CDK_JSON_PATH.read_text(encoding="utf-8")) app_cmd = data.get("app", "") if app_cmd: python_exe = Path(app_cmd.split()[0]) except (json.JSONDecodeError, IndexError): pass values: Dict[str, str] if args.deploy_only: if not ENV_PATH.is_file(): raise SystemExit( f"No config at {ENV_PATH}. Run without --deploy-only first." ) values = read_env_file(ENV_PATH) pre_deploy_errors = validate_globally_unique_env_values( values ) + validate_new_vpc_cidr_env_values(values) if pre_deploy_errors: print("Pre-deploy configuration conflicts:") for err in pre_deploy_errors: print(f" - {err}") return 1 if not python_exe or not python_exe.is_file(): python_exe = resolve_python_executable( override=args.python, interactive=False, assume_yes=True ) else: answers = run_wizard(args) errors = validate_install_answers(answers) errors.extend(validate_subnet_answers(answers)) errors.extend(enrich_existing_subnet_details_from_aws(answers)) values = build_env_values(answers) errors.extend(validate_env_values(values)) if errors: print("Configuration errors:") for err in errors: print(f" - {err}") return 1 if not python_exe: python_exe = resolve_python_executable( override=answers.python_path or args.python, interactive=not args.yes, assume_yes=args.yes, ) if not args.skip_cdk_json: write_cdk_json(python_exe, force=args.refresh_cdk_json) print_summary(values, python_exe) if not args.yes and not ask_yes_no("Write config/cdk_config.env?", True): print("Aborted.") return 0 write_env_file(ENV_PATH, values) if answers.write_app_config_env: write_app_config_env_file( answers, values, overwrite=answers.overwrite_app_config_env, ) apply_cdk_runtime_env(values) if args.config_only: print("Config written. Exiting (--config-only).") return 0 run_smoke_test_if_needed(python_exe, args) # Bootstrap prompt account = values.get("AWS_ACCOUNT_ID", "") region = values.get("AWS_REGION", "") if account and region: try: if cdk_bootstrap_needed(account, region): print(f"CDK bootstrap not found in {region}.") if args.yes or ask_yes_no( f"Run cdk bootstrap aws://{account}/{region}?", True ): run_cdk_command( ["bootstrap", f"aws://{account}/{region}"], check=True, ) except Exception as exc: print(f"Bootstrap check skipped: {exc}") verify_node_for_jsii() run_cdk_command(["synth"], check=True) if args.synth_only or args.skip_deploy: print("Synth complete. Skipping deploy.") return 0 if not args.yes and not ask_yes_no("Run cdk deploy --all?", True): print("Deploy skipped.") return 0 run_cdk_command(["deploy", "--all", "--require-approval", "broadening"], check=True) values = read_env_file(ENV_PATH) apply_post_deploy_fixup(values, assume_yes=args.yes) run_qs = False if args.skip_quickstart: print("Skipping post-deploy quickstart.") else: is_headless = values.get("ENABLE_HEADLESS_DEPLOYMENT") == "True" is_demo = values.get("USE_ECS_EXPRESS_MODE") == "True" if args.yes: run_qs = True elif is_headless: run_qs = ask_yes_no( "Run post_cdk_build_quickstart.py (CodeBuild image only; no ECS service)?", default=True, ) else: run_qs = ask_yes_no( "Run post_cdk_build_quickstart.py (CodeBuild + scale ECS)?", default=is_demo, ) if run_qs: if not python_exe: python_exe = resolve_python_executable( assume_yes=True, interactive=False ) run_quickstart(python_exe) values = read_env_file(ENV_PATH) is_headless = values.get("ENABLE_HEADLESS_DEPLOYMENT") == "True" if values.get("USE_ECS_EXPRESS_MODE") == "True" and not is_headless: from cdk_post_deploy import print_express_mode_next_steps if args.skip_quickstart or not run_qs: print_express_mode_next_steps(values) return 0 if is_headless: from cdk_post_deploy import ( print_headless_deployment_next_steps, print_headless_output_notification_steps, provision_headless_output_reader_access_key, ) if args.skip_quickstart or not run_qs: print_headless_deployment_next_steps(values) if values.get("ENABLE_HEADLESS_OUTPUT_NOTIFICATIONS") == "True": print_headless_output_notification_steps(values) create_key = False if args.skip_headless_output_access_key: create_key = False elif args.create_headless_output_access_key: create_key = True elif not args.yes and not args.deploy_only: create_key = ask_yes_no( "Create IAM access key for the headless output reader user now?", default=True, ) elif args.yes: create_key = True if create_key: provision_headless_output_reader_access_key(values) return 0 print("\nDone. Next steps:") if values.get("USE_CLOUDFRONT") == "True": domain = values.get("SSL_CERTIFICATE_DOMAIN", "") cf = values.get("CLOUDFRONT_DOMAIN", "") if domain and cf and cf != "cloudfront_placeholder.net": print(f" - Point DNS CNAME {domain} -> {cf}") elif values.get("USE_ECS_EXPRESS_MODE") == "True": ep = values.get("ECS_EXPRESS_COGNITO_REDIRECT_BASE", "") if ep: from cdk_config import normalize_https_redirect_url print(f" - Express endpoint: {normalize_https_redirect_url(ep)}") if APP_CONFIG_ENV_PATH.is_file(): print(f" - App runtime config: {APP_CONFIG_ENV_PATH} (uploaded by quickstart)") print(f" - Config: {ENV_PATH}") return 0 if __name__ == "__main__": raise SystemExit(main())