        command.append("-n")
    if extra:
        command.append(extra)
    return " ".join(command)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Clone daylily-omics-analysis and launch a workflow inside tmux.",
    )
    parser.add_argument("--profile", default=os.environ.get("AWS_PROFILE"))
    parser.add_argument("--region", help="AWS region for the cluster")
    parser.add_argument("--cluster", help="ParallelCluster name")
    parser.add_argument(
        "--stage-dir",
        help="Specific staging directory containing *_samples.tsv and *_units.tsv",
    )
    parser.add_argument(
        "--run-context-file",
        help="Local runs.tsv file to write as config/runs.tsv for run-analysis workflows",
    )
    parser.add_argument(
        "--samples-file",
        help="Local samples.tsv file to write as config/samples.tsv for sample-analysis workflows",
    )
    parser.add_argument(
        "--units-file",
        help="Local units.tsv file to write as config/units.tsv for sample-analysis workflows",
    )
    parser.add_argument(
        "--stage-base",
        default="/fsx/staging/staged_external_sequencing_data",
        help="Base staging directory to scan when --stage-dir is omitted",
    )
    parser.add_argument(
        "--no-input-staging",
        dest="input_staging",
        action="store_false",
        help="Do not copy staged samples/units or write a run context before launching",
    )
    parser.add_argument(
        "--no-default-activation",
        dest="default_activation",
        action="store_false",
        help="Do not run the standard dyoainit plus Slurm activation before --dy-command",
    )
    parser.add_argument(
        "--bootstrap-test-config",
        action="store_true",
        help="Copy DayOA bundled test samples and units into config/ before launch",
    )
    parser.add_argument(
        "--session-name",
        help="Name of the tmux session to create on the head node. Defaults to --analysis-id.",
    )
    parser.add_argument(
        "--analysis-id",
        required=True,
        help="Analysis identifier passed to day-clone -d and used under /fsx/analysis_results.",
    )
    parser.add_argument(
        "--executing-entity",
        "-u",
        help="User or system identifier used under /fsx/analysis_results. Defaults to --cluster.",
    )
    parser.add_argument(
        "--repository",
        default="daylily-omics-analysis",
        help="Repository key to pass to day-clone",
    )
    parser.add_argument(
        "--git-tag",
        "-t",
        default="main",
        help="Git branch or tag to pass to day-clone",
    )
    parser.add_argument("--project", help="Project/budget to supply to dyoainit")
    parser.add_argument(
        "--skip-project-check",
        dest="skip_project_check",
        action="store_true",
        help="Skip upstream project validation in dyoainit (default for the supported flow)",
    )
    parser.add_argument(
        "--strict-project-check",
        dest="skip_project_check",
        action="store_false",
        help="Enable upstream project validation in dyoainit",
    )
    parser.add_argument("--genome", default="hg38")
    parser.add_argument("--jobs", type=int, default=6)
    parser.add_argument("--aligners", default="bwa2a")
    parser.add_argument("--dedupers", default="dmd")
    parser.add_argument("--snv-callers", default="deep")
    parser.add_argument("--sv-callers", default="")
    parser.add_argument("--target", default="produce_snv_concordances")
    parser.add_argument("--dy-command", help="Override the dy-r command entirely")
    parser.add_argument("--snakemake-extra", help="Additional arguments appended to dy-r")
    parser.add_argument(
        "--no-containerized",
        action="store_true",
        help="Disable DAY_CONTAINERIZED (enabled by default)",
    )
    parser.add_argument(
        "--export-destination-s3-uri",
        help="Full S3 prefix ending in <executing-entity>/<analysis-id>/ for auto-export",
    )
    parser.add_argument(
        "--export-trigger",
        choices=("none", "on-success", "on-fail", "all"),
        default="none",
        help="Auto-export trigger after the workflow exits",
    )
    parser.add_argument(
        "--delete-on-export-success",
        action="store_true",
        help="Delete the FSx analysis directory after a successful requested export",
    )
    parser.add_argument(
        "--replace-existing-analysis-dir",
        action="store_true",
        help=(
            "Explicit retry mode: remove an existing same analysis directory before "
            "launching. Without this flag, existing analysis directories fail hard."
        ),
    )
    parser.add_argument(
        "--artifact-registration-command-id",
        default="",
        help="Catalog command id whose artifact_registration policy should run after export",
    )
    parser.add_argument(
        "--dewey-url", default="", help="Dewey base URL for post-export registration"
    )
    parser.add_argument(
        "--dewey-token-env",
        default="",
        help="Environment variable containing the Dewey bearer token",
    )
    parser.add_argument(
        "--dewey-analysis-dir-external-object-id",
        default="",
        help="External object id for the exported daylily-omics-analysis S3 directory",
    )
    parser.add_argument(
        "--dewey-run-artifact-euid",
        default="",
        help="Dewey run artifact EUID linked to the exported analysis directory external object",
    )
    parser.add_argument(
        "--dewey-ursa-analysis-euid",
        default="",
        help="Ursa analysis EUID linked to the exported analysis directory external object",
    )
    parser.add_argument("--dry-run", action="store_true")
    parser.set_defaults(skip_project_check=True, input_staging=True, default_activation=True)
    return parser


def validate_export_registration_args(args: argparse.Namespace) -> None:
    if args.export_destination_s3_uri and args.export_trigger == "none":
        raise CommandError("--export-trigger must not be none when auto-export is requested.")
    if args.export_trigger != "none" and not args.export_destination_s3_uri:
        raise CommandError("--export-destination-s3-uri is required when --export-trigger is set.")
    if args.delete_on_export_success and not args.export_destination_s3_uri:
        raise CommandError("--delete-on-export-success requires --export-destination-s3-uri.")
    if args.artifact_registration_command_id and args.export_trigger == "none":
        raise CommandError("--artifact-registration-command-id requires an export trigger.")
    if args.artifact_registration_command_id and not args.dewey_url:
        raise CommandError("--dewey-url is required with --artifact-registration-command-id.")
    if args.artifact_registration_command_id and not args.dewey_token_env:
        raise CommandError("--dewey-token-env is required with --artifact-registration-command-id.")
    if not args.artifact_registration_command_id and (args.dewey_url or args.dewey_token_env):
        raise CommandError(
            "--artifact-registration-command-id is required when Dewey registration options are set."
        )
    dewey_link_options = {
        "--dewey-analysis-dir-external-object-id": args.dewey_analysis_dir_external_object_id,
        "--dewey-run-artifact-euid": args.dewey_run_artifact_euid,
        "--dewey-ursa-analysis-euid": args.dewey_ursa_analysis_euid,
    }
    if any(str(value or "").strip() for value in dewey_link_options.values()):
        missing = [
            option for option, value in dewey_link_options.items() if not str(value or "").strip()
        ]
        if missing:
            raise CommandError(
                "Dewey analysis-directory external-link options must be provided together: "
                + ", ".join(missing)
            )
        if not args.artifact_registration_command_id:
            raise CommandError(
                "--artifact-registration-command-id is required with Dewey external-link options."
            )


def main(argv: Optional[List[str]] = None) -> int:
    args = build_parser().parse_args(argv)

    if not args.profile:
        raise CommandError("AWS profile is required. Set AWS_PROFILE or use --profile.")
    validate_export_registration_args(args)

    need_cmd("aws")
    need_cmd("pcluster")

    region = resolve_region(args.profile, args.region)
    cluster_name = resolve_cluster(args.profile, region, args.cluster)
    analysis_id = validate_analysis_segment(args.analysis_id, field_name="analysis_id")
    executing_entity = validate_analysis_segment(
        args.executing_entity or cluster_name,
        field_name="executing_entity",
    )
    source_path = analysis_source_path(
        executing_entity=executing_entity,
        analysis_id=analysis_id,
        headnode=True,
    )
    if not args.session_name:
        args.session_name = analysis_id
    if args.export_destination_s3_uri:
        from daylily_ec.workflow.export_data import (
            _create_session,
            validate_export_destination_s3_uri,
            validate_s3_destination_prefix_empty,
        )

        validate_export_destination_s3_uri(
            args.export_destination_s3_uri,
            source_path=source_path,
        )
        validate_s3_destination_prefix_empty(
            _create_session(region, args.profile).client("s3"),
            args.export_destination_s3_uri,
            source_path=source_path,
        )
    target = resolve_headnode_instance_id(cluster_name, region, profile=args.profile)
    wait_for_ssm_online(target.instance_id, region, profile=args.profile, timeout=120)
    validate_headnode_readiness(
        target.instance_id,
        region,
        profile=args.profile,
        timeout=120,
        comment="Validate DAY-EC headnode readiness before workflow launch",
    )

    run_context_content: Optional[str] = None
    samples_content: Optional[str] = None
    units_content: Optional[str] = None
    if args.run_context_file:
        if not args.input_staging:
            raise CommandError("--run-context-file cannot be used with --no-input-staging.")
        if args.stage_dir:
            raise CommandError("--stage-dir cannot be used with --run-context-file.")
        if args.samples_file or args.units_file:
            raise CommandError(
                "--run-context-file cannot be used with --samples-file or --units-file."
            )
        run_context_path = Path(args.run_context_file).expanduser()
        if not run_context_path.is_file():
            raise CommandError(f"Run context file not found: {run_context_path}")
        run_context_content = run_context_path.read_text(encoding="utf-8")
        stage_config = None
    elif args.samples_file or args.units_file:
        if not args.input_staging:
            raise CommandError(
                "--samples-file/--units-file cannot be used with --no-input-staging."
            )
        if args.stage_dir:
            raise CommandError("--stage-dir cannot be used with --samples-file/--units-file.")
        if not args.samples_file or not args.units_file:
            raise CommandError("--samples-file and --units-file must be provided together.")
        samples_path = Path(args.samples_file).expanduser()
        units_path = Path(args.units_file).expanduser()
        if not samples_path.is_file():
            raise CommandError(f"Samples file not found: {samples_path}")
        if not units_path.is_file():
            raise CommandError(f"Units file not found: {units_path}")
        samples_content = samples_path.read_text(encoding="utf-8")
        units_content = units_path.read_text(encoding="utf-8")
        stage_config = None
    elif args.input_staging:
        stage_config = discover_stage_config(
            target.instance_id,
            args.profile,
            region,
            args.stage_dir,
            args.stage_base,
        )
    else:
        if args.stage_dir:
            raise CommandError("--stage-dir cannot be used with --no-input-staging.")
        stage_config = None

    if args.dy_command:
        dy_command = args.dy_command
    else:
        dy_command = build_default_command(
            target=args.target,
            genome=args.genome,
            jobs=args.jobs,
            aligners=args.aligners.split(","),
            dedupers=args.dedupers.split(","),
            snv_callers=args.snv_callers.split(","),
            sv_callers=[value for value in args.sv_callers.split(",") if value],
            containerized=not args.no_containerized,
            dry_run=args.dry_run,
            extra=args.snakemake_extra,
        )

    project_arg = shlex.quote(args.project) if args.project else ""
    repository_literal = json.dumps(args.repository)
    dy_command_literal = shlex.quote(dy_command)
    skip_check = "true" if args.skip_project_check else "false"
    run_context_mode = run_context_content is not None
    sample_config_mode = samples_content is not None or units_content is not None
    run_context_mode_literal = "true" if run_context_mode else "false"
    sample_config_mode_literal = "true" if sample_config_mode else "false"
    input_staging_mode_literal = "true" if args.input_staging else "false"
    default_activation_literal = "true" if args.default_activation else "false"
    bootstrap_test_config_literal = "true" if args.bootstrap_test_config else "false"
