band-aider-pub/band_aider/cli.py

# SPDX-FileCopyrightText: 2026 Art Sea LLC
#
# SPDX-License-Identifier: AGPL-3.0-only

"""Command-line entrypoint for the current Band-Aider SupremeLoop."""
from __future__ import annotations

import argparse
import logging
import os
import sys
import time
from multiprocessing import Queue
from pathlib import Path
from typing import Any

from band_aider.judgement.prompt_localization import PromptLocalization

# from band_aider.ai_client.semantic_completion_cache import CachedCompletionClient as SemanticCache

BAND_AIDER_TERMINAL_UI = "BAND_AIDER_TERMINAL_UI"
BAND_AIDER_SKIP_STARTUP = "BAND_AIDER_SKIP_STARTUP"

DEBUG_ALLOWED_FLAGS = {'true', 'yes', 'y', 't', 'ok', 'local', 'dev'}
POSITIVE_FLAGS = {'1', 'true', 'yes', 't', 'y', 'ok'}


def _debug_enabled() -> bool:
    """Return whether pydevd debug mode is enabled for this process."""

    value = os.getenv("DEBUG")
    return value is not None and value.lower() in DEBUG_ALLOWED_FLAGS


def _skip_startup(args: argparse.Namespace) -> bool:
    """Return whether the interactive startup wizard should be bypassed."""

    return bool(getattr(args, "skip_startup", False)) or os.getenv(
        BAND_AIDER_SKIP_STARTUP, ""
    ).strip().lower() in POSITIVE_FLAGS


if _debug_enabled():
    import pydevd_pycharm

    pydevd_pycharm.settrace('localhost', port=9001, stdout_to_server=True, stderr_to_server=True, suspend=False)

from band_aider.ai_client import llm_logging
from band_aider.ai_client.completion_client import CompletionClient, OpenAIConfig
from band_aider.ai_client.cached_completion_client import CachedCompletionClient
from band_aider.ai_client.cached_completion_client import CompletionCache
from band_aider.ai_client.cycle_jacker import CycleJacker
from band_aider.ai_client.fence_crashing_client import FenceCrashingClient
from band_aider.ai_client.human_feedback_store import HumanFeedbackStore
from band_aider.ai_client.json_array_prompt_reducer import GeneralPromptReducer
from band_aider.ai_client.openai_config_factory import OpenAIConfigFactory
from band_aider.ai_client.openai_proxy_factory import OpenAIProxyFactory
from band_aider.ai_client.token_miser import TokenMiser
from band_aider.ai_client.token_tracker import TokenTracker
from band_aider.ai_client.token_jacker import TokenJacker
from band_aider.context.band_aider_facts import get_band_aider_facts
from band_aider.context.token_budget_state import (
    configure_token_budget_settings,
    get_token_budget_settings,
    persist_token_budget_stats_json,
    restore_token_budget_stats_json,
    token_breakdown_dtos,
    token_budget_stats_dto,
)
from band_aider.context.context_factory import context_factory
from band_aider.context.global_context import GlobalContext, SovereignState
from band_aider.judgement.cached_code_editor import CachedCodeEditor
from band_aider.judgement.cached_file_patcher import CachedFilePatcher
from band_aider.judgement.code_edit_or_leave_it import CodeEditOrLeaveIt
from band_aider.judgement.direct_code_editor import DirectCodeEditor
from band_aider.judgement.file_patcher import FilePatcher
from band_aider.judgement.intent_check import IntentCheckFactory
from band_aider.judgement.prompt_formatter import (
    MarkdownPromptFormatterBuilder,
    PromptFormatterBuilder,
    format_prompt,
)
from band_aider.judgement.search_replace_token_generator import SearchReplaceTokenGenerator
from band_aider.loop.sovereign_loop import SovereignLoop
from band_aider.strategies import language_strategy_set_factory
from band_aider.strategies.registration import register_strategies
from band_aider.ui.queue_logging_handler import QueueLoggingHandler
from band_aider.ui.queue_status_publisher import QueueHumanCommandReceiver, QueueStatusPublisher
from band_aider.ui import intervention_selection
from band_aider.ui.dto.human_intervention import CycleInterruptCommandDTO
from band_aider.ai_client.cycle_jacker import cycle
from band_aider.ui.status import set_status_publisher
from band_aider.ui.status import publish_supreme_status
from band_aider.ui.dto.run_completion import RunCompletionDTO
from band_aider.ui.status_publisher import (
    ConsoleHumanInterventionAdapter,
    DeferredHumanCommandReceiver,
    HumanCommandReceiver,
    NullStatusPublisher,
    StatusPublisher,
)
from band_aider.ui.startup_wizard import StartupWizard
from band_aider.ui.stream_capture import QueueTextWriter
from band_aider.ui.terminal_ui import TerminalUI
from band_aider.ui.trace_publisher import TracePublisher
from band_aider.ui.worker_process import start_worker_process
from band_aider.context.resume_state import (
    configure_resume_settings,
    resume_settings,
)
from band_aider.context.aider_config import AiderConfig, AiderConfigFactory
from band_aider.judgement.repo_exec import (
    DEFAULT_REPO_EXEC,
    DockerContainerRepoExec,
    HostRepoExec,
    RepoExec,
    RepoExecError,
    resolve_repo_exec,
)

REPO_CHECKPOINT_DIRNAME = ".band-aider-chkpts"
# Retained as a compatibility seam for callers/tests that used to redirect the
# removed host-side last-project pointer. It is intentionally never read or written.
LAST_REPO_ROOT_PATH: Path | None = None


def build_parser() -> argparse.ArgumentParser:
    """Build the Band-Aider CLI parser."""

    parser = argparse.ArgumentParser(prog="band-aider")
    parser.add_argument("--task")
    parser.add_argument("--language", default="python")
    parser.add_argument("--src-root", default="band_aider")
    parser.add_argument("--tests-root", default="tests")
    parser.add_argument("--actor", default="band-aider")
    parser.add_argument("--dry-run", action="store_true")
    parser.add_argument("--localize-prompts", action="store_true",
                        help="Opt into experimental prompt localization (disabled by default).")
    parser.add_argument("--repo-root")
    parser.add_argument("--resume", action="store_true")
    parser.add_argument("--aider-cmd")
    parser.add_argument("--aider-model")
    parser.add_argument("--aider-api-base")
    parser.add_argument("--aider-api-key")
    parser.add_argument("--aider-edit-format")
    parser.add_argument("--aider-wrapper-mode", nargs="?", const="true")
    parser.add_argument("--aider-extra-arg", action="append", default=[])
    parser.add_argument("--model")
    parser.add_argument("--model-api-base")
    parser.add_argument("--model-api-key")
    parser.add_argument("--model-ignore-ssl", action="store_true",
                        help="Disable SSL certificate verification for the primary model endpoint.")
    parser.add_argument("--max-prompt-tokens", type=int)
    parser.add_argument("--tokens-per-word", type=float)
    parser.add_argument("--token-overflow-strategy", choices=("miser", "jacker"))
    parser.add_argument("--max-output-tokens", type=int)
    parser.add_argument("--terminal-ui", action="store_true")
    parser.add_argument("--intervention", action="store_true", help="Request one manual stop at the next matching interaction.")
    parser.add_argument("--intervention-setup", action="store_true", help="Choose intervention targets with prefix and Tab completion before running.")
    parser.add_argument("--intervention-decision", default="", help="Concrete decision name or unique prefix; blank matches any.")
    parser.add_argument("--intervention-file", default="", help="Exact repository-relative focused/editable path; blank matches any.")
    parser.add_argument("--intervention-response", choices=("guide", "replace"), default="guide")
    parser.add_argument("--log-to-console", action="store_true")
    parser.add_argument(
        "--trace-file",
        default="",
        help="Append every status event to this JSONL trace file "
        "(a relative path resolves against the invocation directory).",
    )
    parser.add_argument(
        "--skip-startup",
        action="store_true",
        help="Skip the interactive splash/startup wizard.",
    )
    parser.add_argument(
        "--repo-container",
        help="Running Docker container name or ID for persistent container-backed execution",
    )
    parser.add_argument(
        "--repo-python-bin",
        help="Python executable inside --repo-container (or BAND_AIDER_DOCKER_PYTHON_BIN)",
    )
    parser.add_argument("--worker-mode", action="store_true", help=argparse.SUPPRESS)
    parser.add_argument(
        "--confirm-empty-file-cleanup",
        action="store_true",
        help="Delete zero-byte files under source root and tests-root before running.",
    )
    return parser


def _health_check(
    client: CompletionClient,
    prompt_formatter_builder: PromptFormatterBuilder,
) -> None:
    """Verify the configured model client can answer a trivial readiness prompt."""

    response = client.complete(
        format_prompt(
            prompt_formatter_builder,
            "MODEL HEALTH CHECK",
            "Reply exactly: BAND_AIDER_READY",
        )
    )
    if "BAND_AIDER_READY" not in response:
        raise RuntimeError(f"Unexpected health check response: {response!r}")


def _model_is_configured(client: CompletionClient) -> bool:
    """Return whether the model client has enough configuration for normal execution."""

    return bool(
        getattr(client, "model", "")
        and getattr(client, "api_base", "")
        and getattr(client, "api_key", "")
    )


def _overflow_strategy(args: argparse.Namespace) -> str:
    """Return the configured overflow strategy."""

    return (
            args.token_overflow_strategy
            or os.getenv("BAND_AIDER_TOKEN_OVERFLOW_STRATEGY")
            or "miser"
    ).strip().lower()


def _terminal_ui(args: argparse.Namespace) -> bool:
    """Retruns True if terminal ui is enabled, False otherwise"""
    return args.terminal_ui or os.getenv(BAND_AIDER_TERMINAL_UI, "").strip().lower() in POSITIVE_FLAGS


def _max_output_tokens(args: argparse.Namespace) -> int:
    """Return the configured completion output cap."""

    settings = get_token_budget_settings()
    if args.max_output_tokens is not None:
        return args.max_output_tokens
    env_value = os.getenv("BAND_AIDER_MAX_OUTPUT_TOKENS")
    if env_value:
        return int(env_value)
    return settings.max_output_tokens


def _runtime_client(
        client: CompletionClient,
        args: argparse.Namespace,
        configs: tuple[AiderConfig, ...],
        prompt_formatter_builder: PromptFormatterBuilder,
        repo_root: Path = Path("."),
        repo_exec: RepoExec = DEFAULT_REPO_EXEC,
        event_publisher: StatusPublisher | None = None,
        command_receiver: HumanCommandReceiver | None = None,
        cycle_detection: bool = False,
        config_index: int = 0,
        config_count: int | None = None,
        escalation_clients: tuple[CompletionClient, ...] = (),
) -> CompletionClient:
    """Wrap the model client with response cleanup and prompt token budgeting."""

    repo_exec = resolve_repo_exec(repo_exec)
    strategy = _overflow_strategy(args)
    guarded: CompletionClient
    if strategy == "jacker":
        guarded = TokenJacker(
            inner=client,
            settings=get_token_budget_settings(),
            event_publisher=event_publisher,
            escalation_clients=escalation_clients,
        )
    else:
        guarded = TokenMiser(
            inner=client,
            settings=get_token_budget_settings(),
            reducers=[GeneralPromptReducer()],
        )
    cleaned: CompletionClient = FenceCrashingClient(inner=guarded)
    tracked: CompletionClient
    if not cycle_detection:
        tracked = cleaned
    else:
        tracked = CycleJacker(
            inner=cleaned,
            event_publisher=event_publisher or NullStatusPublisher(),
            command_receiver=command_receiver or DeferredHumanCommandReceiver(),
            feedback_store=HumanFeedbackStore(
                repo_root,
                repo_exec=repo_exec,
            ),
            escalation_clients=escalation_clients,
            prompt_formatter_builder=prompt_formatter_builder,
        )
    return TokenTracker(
        inner=tracked,
        config_index=config_index,
        config_count=config_count if config_count is not None else len(configs),
        event_publisher=event_publisher,
        command_receiver=command_receiver,
    )


def _runtime_clients(
        openai_configs: tuple[OpenAIConfig, ...],
        args: argparse.Namespace,
        configs: tuple[AiderConfig, ...],
        prompt_formatter_builder: PromptFormatterBuilder,
        repo_root: Path = Path("."),
        repo_exec: RepoExec = DEFAULT_REPO_EXEC,
        event_publisher: StatusPublisher | None = None,
        command_receiver: HumanCommandReceiver | None = None,
        cycle_detection: bool = False,
) -> tuple[CompletionClient, ...]:
    """Return runtime clients aligned to the configured escalation order.

    Every caller uses the same cached completion stack.  Tool handlers own
    idempotency against their selected repository resource, so replayed model
    responses cannot duplicate a mutation.
    """

    repo_exec = resolve_repo_exec(repo_exec)
    runtime_clients: list[CompletionClient] = []
    raw_clients = OpenAIProxyFactory.from_configs(openai_configs)
    for index, client in enumerate(raw_clients):
        if hasattr(client, "extra_openai_configs"):
            client.extra_openai_configs = openai_configs[index + 1:]
    model_clients = tuple(
        CachedCompletionClient(client, cache=CompletionCache())
        for client in raw_clients
    )
    for index, model_client in enumerate(model_clients):
        runtime_clients.append(
            _runtime_client(
                model_client,
                args,
                configs[index:],
                repo_root=repo_root,
                repo_exec=repo_exec,
                event_publisher=event_publisher,
                command_receiver=command_receiver,
                cycle_detection=cycle_detection,
                config_index=index,
                config_count=len(openai_configs),
                escalation_clients=model_clients[index + 1:],
                prompt_formatter_builder=prompt_formatter_builder,
            )
        )
    return tuple(runtime_clients)


def _run_worker(argv: list[str], event_queue: Queue, command_queue: Queue) -> None:
    """Run the CLI workflow in a child process with queue-backed UI adapters."""

    sys.stdout = QueueTextWriter(event_queue, "worker.stdout")
    sys.stderr = QueueTextWriter(event_queue, "worker.stderr", level="WARN")
    args = [*argv, "--worker-mode"]
    raise SystemExit(main(args, event_queue=event_queue, command_queue=command_queue))


def _client_model_name(model: str) -> str:
    """Return the downstream OpenAI-compatible payload model name."""

    return model[7:] if model.startswith("openai/") else model


def _aider_model_name(model: str) -> str:
    """Return the provider-qualified model name shown in current CLI diagnostics."""

    return model if "/" in model else f"openai/{model}"


def _path_value(value: Any, repo_exec: RepoExec) -> Path | None:
    """Return one checkpoint coordinate normalized by the active execution adapter."""

    if isinstance(value, Path):
        return Path(repo_exec.path_in_exec(value))
    if isinstance(value, str) and value.strip():
        return Path(repo_exec.path_in_exec(Path(value)))
    return None


def _payload_roots(
        payload: dict[str, Any],
        *,
        repo_exec: RepoExec,
) -> tuple[Path, Path, Path] | None:
    """Return repo/src/test roots from a checkpoint payload when all are present."""

    manager = payload.get("_file_manager")
    if manager is not None:
        repo_root = _path_value(getattr(manager, "repo_root", None), repo_exec)
        src_root = _path_value(getattr(manager, "src_root", None), repo_exec)
        test_root = _path_value(getattr(manager, "test_root", None), repo_exec)
        if repo_root and src_root and test_root:
            return repo_root, src_root, test_root

    repo_root = _path_value(payload.get("repo_root"), repo_exec)
    src_root = _path_value(payload.get("src_root"), repo_exec)
    test_root = _path_value(payload.get("test_root"), repo_exec)
    if repo_root and src_root and test_root:
        return repo_root, src_root, test_root
    return None


def _effective_roots(args: argparse.Namespace, repo_exec: RepoExec) -> tuple[Path, Path, Path]:
    """Return the roots that should govern this run."""

    explicit_repo_root = args.repo_root is not None
    _validate_relative_root_argument("source", args.src_root)
    _validate_relative_root_argument("test", args.tests_root)
    repo_root = Path(
        repo_exec.path_in_exec(_authoritative_repo_root(args))
    )
    # Path normalization is delegated to the selected execution environment.  In
    # host mode this gives the normal resolved host path; in container mode it
    # remains a lexical container path and never consults the host filesystem.
    src_root = Path(repo_exec.path_in_exec(repo_root / args.src_root))
    test_root = Path(repo_exec.path_in_exec(repo_root / args.tests_root))
    _validate_project_roots(repo_root, src_root, test_root)
    if not args.resume:
        return repo_root, src_root, test_root

    payload = context_factory.latest_payload_for(GlobalContext)
    if payload is None:
        raise FileNotFoundError(
            f"Band-Aider requires an existing GlobalContext checkpoint in {resume_settings().checkpoint_dir} or a --task."
        )
    if explicit_repo_root:
        return repo_root, src_root, test_root

    checkpoint_roots = _payload_roots(payload or {}, repo_exec=repo_exec)
    if checkpoint_roots is None:
        raise ValueError(
            "Band-Aider resume checkpoint is missing repo_root/src_root/test_root coordinates."
        )
    _validate_project_roots(*checkpoint_roots)
    return checkpoint_roots


def _validate_project_roots(
        repo_root: Path,
        src_root: Path,
        test_root: Path,
) -> None:
    """Require disjoint source and test roots contained by the repository root."""

    for label, path in (("source", src_root), ("test", test_root)):
        raw = str(path)
        if Path(raw).is_absolute() and not str(path).startswith(str(repo_root)):
            raise ValueError(f"{label} root must be repository-relative: {path}")
        if any(part == ".." for part in Path(raw).parts):
            raise ValueError(f"{label} root must not contain '..': {path}")
        try:
            path.relative_to(repo_root)
        except ValueError as error:
            raise ValueError(
                f"{label} root must be inside repository root: {path}"
            ) from error

    if src_root == test_root:
        raise ValueError("Source and test roots must not overlap.")
    try:
        src_root.relative_to(test_root)
    except ValueError:
        pass
    else:
        raise ValueError("Source and test roots must not overlap.")
    try:
        test_root.relative_to(src_root)
    except ValueError:
        return
    raise ValueError("Source and test roots must not overlap.")


def _validate_relative_root_argument(label: str, value: str) -> None:
    """Reject non-relative or traversing public project coordinates before IO."""
    candidate = Path(value)
    if candidate.is_absolute() or (len(value) >= 2 and value[1] == ":"):
        raise ValueError(f"{label} root must be relative to --repo-root: {value}")
    if any(part == ".." for part in candidate.parts):
        raise ValueError(f"{label} root must not contain '..': {value}")
    if value in {"", "."}:
        raise ValueError(f"{label} root must name a proper repository subdirectory: {value!r}")


def _authoritative_repo_root(args: argparse.Namespace) -> Path:
    """Return the authoritative repository root for this run."""

    # Project selection is intentionally never persisted in host temporary state.
    # Container runs must receive the container and lexical repository root again.

    if args.repo_root:
        # In container mode, do not resolve against the host filesystem
        return Path(str(args.repo_root))

    if getattr(args, "repo_container", None):
        raise ValueError("Container mode requires --repo-root; host last-project state is disabled.")
    if not sys.stdin.isatty():
        raise ValueError(
            "Band-Aider requires --repo-root or interactive repository selection."
        )

    response = input("Repository root: ").strip()
    if not response:
        raise ValueError("Band-Aider requires a non-empty repository root.")
    return Path(response)


def _checkpoint_dir(repo_root: Path) -> Path:
    """Return the checkpoint directory for one repository root."""

    return repo_root / REPO_CHECKPOINT_DIRNAME


def _has_save_points(repo_root: Path, repo_exec: RepoExec) -> bool:
    """Return whether a repository root appears to have saved Band-Aider state."""

    checkpoint_dir = _checkpoint_dir(repo_root)
    if not repo_exec.is_dir(checkpoint_dir):
        return False
    return bool(repo_exec.list_directory(checkpoint_dir))


def _yes_no(prompt: str, default: bool = False) -> bool:
    """Ask a yes/no question and return the selected answer."""

    suffix = " [Y/n]: " if default else " [y/N]: "
    response = input(f"{prompt}{suffix}").strip().lower()
    if not response:
        return default
    return response in {"y", "yes"}


def _terminal_ui_repo_root(args: argparse.Namespace) -> Path:
    """Resolve or ask for the repository root before launching the UI worker."""

    if args.repo_root:
        return Path(str(args.repo_root))

    if not sys.stdin.isatty():
        raise ValueError(
            "Band-Aider terminal UI requires --repo-root when no interactive repository root is supplied."
        )

    response = input("Repository root: ").strip()
    if not response:
        raise ValueError("Band-Aider requires a non-empty repository root.")
    return Path(response)


def _terminal_ui_task() -> str:
    """Ask for the task text before launching a new UI worker run."""

    response = input("Task: ").strip()
    if not response:
        raise ValueError("Band-Aider requires a non-empty task.")
    return response


def _strip_worker_start_options(argv: list[str]) -> list[str]:
    """Remove parent-owned startup options before constructing worker argv."""

    stripped: list[str] = []
    skip_next = False
    value_options = {"--repo-root", "--task", "--token-overflow-strategy", "--language"}
    flag_options = {"--terminal-ui", "--skip-startup", "--worker-mode", "--resume", "--log-to-console"}
    for item in argv:
        if skip_next:
            skip_next = False
            continue
        if item in value_options:
            skip_next = True
            continue
        if any(item.startswith(f"{option}=") for option in value_options):
            continue
        if item in flag_options:
            continue
        stripped.append(item)
    return stripped


def _terminal_ui_worker_argv(args: argparse.Namespace, argv: list[str]) -> list[str]:
    """Return worker arguments after parent-owned terminal UX prompting."""

    if args.language not in language_strategy_set_factory.keys():
        raise ValueError(
            "Band-Aider language is not available yet: "
            f"{args.language}. Supported languages: "
            f"{', '.join(language_strategy_set_factory.supported_languages())}"
        )

    # Always launch the StartupWizard when running in an interactive terminal.
    # The wizard prompts the user and its choices override CLI args, while
    # defaulting to any provided CLI/env values.
    if sys.stdin.isatty() and sys.stdout.isatty() and not _skip_startup(args):
        initial_repo_root = Path(args.repo_root) if args.repo_root else None

        def resume_available(root: Path) -> bool:
            """Check wizard-selected checkpoint state through the project IO adapter."""

            container_id = getattr(args, "repo_container", None)
            selected_repo_exec = (
                DockerContainerRepoExec(
                    container=container_id,
                    repo_root=root,
                    python_bin=getattr(args, "repo_python_bin", None),
                )
                if container_id
                else HostRepoExec(root)
            )
            return selected_repo_exec.is_dir(root) and _has_save_points(root, selected_repo_exec)

        initial_resume = False
        if initial_repo_root is not None and args.resume and not args.task:
            initial_resume = resume_available(initial_repo_root)
        try:
            startup = StartupWizard(
                initial_repo_root=initial_repo_root,
                initial_task=args.task,
                initial_resume=initial_resume,
                initial_overflow_strategy=_overflow_strategy(args),
                initial_language=args.language,
                resume_available=resume_available,
            ).run()
        except Exception:
            startup = None
        if startup is not None:
            repo_root = startup.repo_root
            resume = startup.resume
            task = startup.task
            token_overflow_strategy = getattr(startup, "token_overflow_strategy", None)
            language = getattr(startup, "language", args.language)
        else:
            repo_root = _terminal_ui_repo_root(args)
            resume = args.resume
            task = args.task
            token_overflow_strategy = None
            language = args.language
    else:
        repo_root = _terminal_ui_repo_root(args)
        resume = args.resume
        task = args.task
        token_overflow_strategy = None
        language = args.language

    container_id = getattr(args, "repo_container", None)
    repo_exec = (
        DockerContainerRepoExec(
            container=container_id,
            repo_root=repo_root,
            python_bin=getattr(args, "repo_python_bin", None),
        )
        if container_id
        else HostRepoExec(repo_root)
    )
    if not repo_exec.is_dir(repo_root):
        raise ValueError(f"Repository root does not exist: {repo_root}")
    save_points_exist = _has_save_points(repo_root, repo_exec)
    if resume and not save_points_exist:
        raise ValueError(f"No saved Band-Aider state found for repository root: {repo_root}")
    if not resume and not task:
        if save_points_exist:
            resume = _yes_no("Saved Band-Aider state was found. Resume it?", default=True)
        if not resume:
            task = _terminal_ui_task()

    worker_argv = _strip_worker_start_options(argv)
    worker_argv.extend(["--repo-root", str(repo_root)])
    if resume:
        worker_argv.append("--resume")
    else:
        worker_argv.extend(["--task", task or _terminal_ui_task()])
    # Thread through overflow strategy selected in the wizard, if present
    if token_overflow_strategy:
        worker_argv.extend(["--token-overflow-strategy", token_overflow_strategy])
    if language:
        worker_argv.extend(["--language", language])
    return worker_argv


def _configure_runtime_resume_settings(
        args: argparse.Namespace,
        repo_root: Path,
        repo_exec: RepoExec,
) -> None:
    """Bind singleton resume settings to the authoritative repo root for this process."""

    configure_resume_settings(
        hydrate=args.resume,
        checkpoint_dir=repo_root / REPO_CHECKPOINT_DIRNAME,
        repo_exec=repo_exec,
    )


def _empty_files_under(repo_root: Path, scan_root: Path, repo_exec: RepoExec) -> list[Path]:
    """Return zero-byte files under one root, skipping hidden and dunder paths."""

    skipped_dirs = {
        ".git",
        "__pycache__",
        ".pytest_cache",
        ".mypy_cache",
        ".ruff_cache",
        "node_modules",
    }
    tracked: set[str] = set()
    tracked_result = repo_exec.run_cli(
        ["git", "ls-files", "-z"],
        cwd=repo_root,
        capture_output=True,
        text=True,
        check=False,
    )
    if tracked_result.returncode == 0:
        tracked = {
            item for item in (tracked_result.stdout or "").split("\x00") if item
        }
    empty_files: list[Path] = []
    for candidate in repo_exec.list_files(scan_root, ("*",), recursive=True):
        relative_parts = candidate.relative_to(scan_root).parts if candidate != scan_root else ()
        if any(part in skipped_dirs or part.startswith(".") or _is_dunder_name(part) for part in relative_parts):
            continue
        repo_relative = candidate.relative_to(repo_root).as_posix()
        if repo_relative in tracked:
            continue
        try:
            if repo_exec.read_bytes(candidate) == b"":
                empty_files.append(candidate)
        except OSError:
            continue
    return sorted(empty_files)


def _is_dunder_name(name: str) -> bool:
    """Return whether one basename is wrapped in double underscores."""

    stem = Path(name).stem
    return len(stem) > 4 and stem.startswith("__") and stem.endswith("__")


def _confirm_empty_file_cleanup(empty_files: list[Path]) -> bool:
    """Ask the user whether zero-byte files should be deleted."""

    print("Band-Aider found zero-byte files that can corrupt the workflow:")
    for path in empty_files:
        print(f"  {path}")
    response = input("Delete these empty files before continuing? [y/N]: ")
    return response.strip().lower() in {"y", "yes"}


def _cleanup_empty_files(args: argparse.Namespace, repo_root: Path, scan_root: Path, repo_exec: RepoExec) -> int:
    """Delete empty files in the source workflow root after explicit confirmation."""

    empty_files = _empty_files_under(repo_root, scan_root, repo_exec)
    if not empty_files:
        return 0

    if args.confirm_empty_file_cleanup:
        confirmed = True
    elif sys.stdin.isatty():
        confirmed = _confirm_empty_file_cleanup(empty_files)
    else:
        print(
            "Band-Aider found zero-byte files under the source root. "
            "Re-run with --confirm-empty-file-cleanup to delete them before execution.",
            file=sys.stderr,
        )
        for path in empty_files:
            print(path, file=sys.stderr)
        return 2

    if not confirmed:
        print("Band-Aider aborted because empty-file cleanup was not confirmed.", file=sys.stderr)
        return 2

    for path in empty_files:
        repo_exec.delete(path, missing_ok=True)
    print(f"Deleted {len(empty_files)} zero-byte file(s) before starting the workflow.")
    return 0


def _print_dry_run(
        args: argparse.Namespace,
        config_discovery_root: Path,
        src_root: Path,
        test_root: Path,
        config,
        configs,
) -> None:
    """Print the resolved dry-run configuration without secrets."""

    logger = llm_logging.current_logger()
    print("Band-Aider dry run")
    print(f"task: {args.task}")
    print(f"language: {args.language}")
    logger.info(
        "supported languages: "
        f"{', '.join(language_strategy_set_factory.supported_languages())}"
    )
    print(f"aider config discovery root: {config_discovery_root}")
    print(f"source root: {src_root}")
    print(f"tests root: {test_root}")
    print(f"aider command: {config.aider_cmd}")
    print(f"aider model: {config.model}")
    print(f"aider normalized model: {_aider_model_name(config.model)}")
    print(f"client model: {_client_model_name(config.model)}")
    print(f"api base: {config.api_base}")
    print(f"token overflow strategy: {_overflow_strategy(args)}")
    print(f"max prompt tokens: {get_token_budget_settings().max_prompt_tokens}")
    print(f"max output tokens: {_max_output_tokens(args)}")
    if args.trace_file:
        print(f"trace file: {args.trace_file}")
    if _overflow_strategy(args) == "jacker":
        print(f"token overflow config available: {'yes' if len(configs) > 1 else 'no'}")
    print(f"aider escalation configs: {len(configs)}")
    for index, item in enumerate(configs):
        api_key = "[redacted]" if item.api_key else "none"
        print(
            f"aider config {index}: model={item.model} "
            f"aider_model={_aider_model_name(item.model)} "
            f"api_base={item.api_base or 'default'} api_key={api_key}"
        )


def _human_io_adapters(
        event_queue: Queue | None,
        command_queue: Queue | None,
) -> tuple[StatusPublisher, HumanCommandReceiver]:
    """Return status/input adapters for terminal UI, plain CLI, or unattended execution."""

    if event_queue is not None and command_queue is not None:
        publisher = QueueStatusPublisher(event_queue)
        return publisher, QueueHumanCommandReceiver(command_queue, publisher)
    if event_queue is not None or command_queue is not None:
        raise ValueError("Terminal UI human intervention transport requires both event and command queues.")
    if sys.stdin.isatty():
        adapter = ConsoleHumanInterventionAdapter()
        return adapter, adapter
    return NullStatusPublisher(), DeferredHumanCommandReceiver()


def main(
        argv: list[str] | None = None,
        event_queue: Queue | None = None,
        command_queue: Queue | None = None,
) -> int:
    """Run the Band-Aider SupremeLoop from CLI arguments."""

    # Bind concrete handlers before any language-dependent startup branch.
    register_strategies()
    operation_started = time.monotonic()
    parser = build_parser()
    args = parser.parse_args(argv)
    PromptLocalization.enabled = args.localize_prompts
    args.terminal_ui = _terminal_ui(args)
    # Interactive selector setup requires the plain CLI's sole stdin owner.
    if args.intervention_setup and (not sys.stdin.isatty() or args.terminal_ui or event_queue is not None):
        parser.error("--intervention-setup requires an interactive plain CLI; use explicit --intervention options otherwise.")
    if args.terminal_ui and not args.worker_mode and not _debug_enabled():
        try:
            parent_argv = _terminal_ui_worker_argv(args, list(argv or sys.argv[1:]))
        except ValueError as error:
            print(str(error), file=sys.stderr)
            return 2
        ui_event_queue: Queue = Queue()
        ui_command_queue: Queue = Queue()
        worker = start_worker_process(_run_worker, (parent_argv, ui_event_queue, ui_command_queue))
        return TerminalUI(ui_event_queue, ui_command_queue).run_until_worker_exit(worker)

    if not args.task and not args.resume:
        parser.print_help(sys.stderr)
        return 2
    try:
        repo_root = _authoritative_repo_root(args)
        container_id = getattr(args, "repo_container", None)
        repo_exec: RepoExec = (
            DockerContainerRepoExec(
                container=container_id,
                repo_root=repo_root,
                python_bin=getattr(args, "repo_python_bin", None),
            )
            if container_id
            else HostRepoExec(repo_root)
        )
        repo_root = Path(repo_exec.path_in_exec(repo_root))
        # A container dry run only renders the planned command/configuration;
        # the caller may intentionally provide a container-only path that is
        # unavailable to this host-side process.  Real execution still probes
        # the selected environment through RepoExec.
        if not (args.dry_run and container_id) and not repo_exec.is_dir(repo_root):
            raise ValueError(f"Repository root does not exist: {repo_root}")
        _configure_runtime_resume_settings(args, repo_root, repo_exec)
        repo_root, src_root, test_root = _effective_roots(args, repo_exec)
    except RepoExecError as error:
        print(
            f"Band-Aider cannot access the supplied repository container: {error}. "
            "Band-Aider does not manage that container; repair or restart it and rerun with --resume.",
            file=sys.stderr,
        )
        return 1
    except (FileNotFoundError, ValueError) as error:
        print(str(error), file=sys.stderr)
        return 2
    llm_logging.initialize_llm_logging(
        repo_root,
        repo_exec,
        log_to_console=args.log_to_console and event_queue is None,
    )
    status_publisher, command_receiver = _human_io_adapters(
        event_queue,
        command_queue,
    )
    if args.trace_file:
        try:
            # Wrap before queue logging attaches so log events reach the trace too.
            status_publisher = TracePublisher.at_path(
                status_publisher, Path(args.trace_file).expanduser()
            )
        except OSError as exc:
            parser.error(f"--trace-file is not writable: {exc}")
    if event_queue is not None:
        llm_logging.attach_queue_logging_handler(QueueLoggingHandler(status_publisher))
    set_status_publisher(status_publisher)
    # Each run begins without inherited manual requests, cues, or suppression choices.
    cycle.human_interrupt = False
    cycle.intervention_suppressed = False
    cycle.target = CycleInterruptCommandDTO()
    cycle.cues.clear()
    if args.intervention or args.intervention_decision or args.intervention_file:
        try:
            decision = intervention_selection.resolve_choice(args.intervention_decision, intervention_selection.DECISION_CHOICES)
            selected_file = args.intervention_file
            # Explicit CLI paths share the worker's lexical repository identity, including Docker paths.
            if selected_file:
                path = Path(os.path.normpath(repo_root / selected_file))
                selected_file = path.relative_to(repo_root).as_posix()
                if not (path.is_relative_to(src_root) or path.is_relative_to(test_root)):
                    raise ValueError("Intervention file must belong to the managed source or test roots.")
                if not repo_exec.exists(path):
                    raise ValueError(f"Intervention file does not exist: {selected_file}")
            cycle.arm(CycleInterruptCommandDTO(decision, selected_file, args.intervention_response), status_publisher)
        except ValueError as error:
            parser.error(str(error))
    job = args.task or ""
    configure_token_budget_settings(
        max_prompt_tokens=args.max_prompt_tokens,
        max_output_tokens=args.max_output_tokens,
        tokens_per_word=args.tokens_per_word,
    )
    if not args.resume:
        get_band_aider_facts().reset()
    if args.resume and not args.dry_run:
        restore_token_budget_stats_json(resume_settings().checkpoint_dir, repo_exec=repo_exec)
    get_band_aider_facts().begin_operation(operation_started)
    publish_supreme_status(token_budget=token_budget_stats_dto())
    config_discovery_root = Path.cwd().resolve()
    configs = AiderConfigFactory.sequence(args, discovery_root=config_discovery_root)
    config = configs[0]
    openai_configs = tuple(
        config
        for config in OpenAIConfigFactory.sequence(args, discovery_root=config_discovery_root)
        if config.api_base
    )
    if not openai_configs:
        print(
            "Band-Aider requires model, API base, and API key configuration.",
            file=sys.stderr,
        )
        return 1
    if args.language not in language_strategy_set_factory.keys():
        print(
            "Band-Aider language is not available yet: "
            f"{args.language}. Supported languages: "
            f"{', '.join(language_strategy_set_factory.supported_languages())}",
            file=sys.stderr,
        )
        return 2
    client: CompletionClient = OpenAIProxyFactory.from_config(openai_configs[0])
    if not _model_is_configured(client):
        print(
            "Band-Aider requires model, API base, and API key configuration.",
            file=sys.stderr,
        )
        return 1
    prompt_formatter_builder = MarkdownPromptFormatterBuilder()
    runtime_clients = _runtime_clients(
        openai_configs,
        args,
        configs,
        repo_root=repo_root,
        repo_exec=repo_exec,
        event_publisher=status_publisher,
        command_receiver=command_receiver,
        cycle_detection=not args.dry_run,
        prompt_formatter_builder=prompt_formatter_builder,
    )

    if args.dry_run:
        _print_dry_run(args, config_discovery_root, src_root, test_root, config, configs)
        return 0

    # Build the cross-cutting repository exec adapter used by parser, strategies, and patching
    # Empty-file cleanup must not use host filesystem in container mode
    cleanup_result = _cleanup_empty_files(args, repo_root, src_root, repo_exec)
    if cleanup_result != 0:
        return cleanup_result

    try:
        _health_check(client, prompt_formatter_builder)
    except Exception as error:
        print(f"Band-Aider model health check failed: {error}", file=sys.stderr)
        for line in OpenAIConfigFactory.resolution_report(args, discovery_root=config_discovery_root):
            print(line, file=sys.stderr)
        return 1
    runtime_client: CompletionClient = runtime_clients[0]
    logger = logging.getLogger("band_aider.cli")

    cached_file_patcher = CachedFilePatcher(FilePatcher(repo_root, repo_exec=repo_exec))
    direct_editor = DirectCodeEditor(
        token_generator=SearchReplaceTokenGenerator(
            runtime_clients,
            repo_root,
            repo_exec,
            event_publisher=status_publisher,
            prompt_formatter_builder=prompt_formatter_builder,
        ),
        prompt_formatter_builder=prompt_formatter_builder,
        file_patcher=cached_file_patcher,
        repo_root=repo_root,
    )
    aider_worker = CachedCodeEditor(
        inner=CodeEditOrLeaveIt(
            inner=direct_editor,
            client=runtime_client,
            repo_root=repo_root,
            repo_exec=repo_exec,
            prompt_formatter_builder=prompt_formatter_builder,
        ),
        file_patcher=cached_file_patcher,
    )
    try:
        loop = SovereignLoop(
            client=runtime_client,
            repo_root=repo_root,
            src_root=src_root,
            test_root=test_root,
            actor=args.actor,
            intent_check_factory=IntentCheckFactory(
                runtime_client,
                prompt_formatter_builder,
            ),
            original_prompt=job,
            aider_worker=aider_worker,
            repo_exec=repo_exec,
            language=args.language,
            prompt_formatter_builder=prompt_formatter_builder,
            logger=logger,
        )
        logger.info("[probe] SovereignLoop constructed")
    except Exception:
        logger.exception("Band-Aider SovereignLoop construction traceback")
        raise
    # Inventory discovery completes before optional plain-CLI choice entry.
    if args.intervention_setup and isinstance(command_receiver, ConsoleHumanInterventionAdapter):
        try:
            target = intervention_selection.read_target(command_receiver.files, input, sys.stderr)
        except (EOFError, KeyboardInterrupt):
            return 130
        cycle.arm(target, status_publisher)
    try:
        summary = loop.run()
    except RepoExecError as error:
        logger.error(
            f"Band-Aider could not execute repository operation: {error}. "
            "Band-Aider does not manage the supplied container; repair or restart it "
            "and rerun with --resume.",
        )
        return 1
    except Exception:
        logger.exception("Band-Aider workflow failed before completion.")
        raise
    facts = get_band_aider_facts()
    facts.finish_operation()
    facts.persist(resume_settings().checkpoint_dir, repo_exec=repo_exec)
    completion_state = getattr(
        getattr(loop, "context", None),
        "supreme_completion_state",
        None,
    )
    if (
        completion_state is not None
        and SovereignState.coerce(
            completion_state,
            SovereignState.SUPREME_ACTIVE,
        )
        is SovereignState.SUPREME_ACTIVE
    ):
        logger.error(
            "Band-Aider stopped with Supreme still active; retaining the checkpoint "
            "for resume instead of publishing completion."
        )
        persist_token_budget_stats_json(
            resume_settings().checkpoint_dir,
            repo_exec=repo_exec,
        )
        return 1
    logger.info(summary)
    facts_snapshot = facts.snapshot()
    status_publisher.publish(
        RunCompletionDTO(
            final_summary=summary,
            report_path=facts_snapshot.report_path,
            total_tokens_spent=facts_snapshot.total_tokens_spent,
            largest_query_tokens=facts_snapshot.largest_query_tokens or facts_snapshot.last_query_tokens,
            total_seconds=facts_snapshot.total_seconds,
            average_seconds_per_cot=facts_snapshot.average_seconds_per_cot,
            cot_count=facts_snapshot.cot_count,
            files_modified=facts_snapshot.files_modified,
            files_added=facts_snapshot.files_added,
            llm_call_count=facts_snapshot.llm_call_count,
            llm_wait_seconds=facts_snapshot.llm_wait_seconds,
            baseline_tokens_spent=facts_snapshot.baseline_tokens_spent,
            escalated_tokens_spent=facts_snapshot.escalated_tokens_spent,
            token_breakdown=token_breakdown_dtos(facts_snapshot.token_breakdown),
            total_cache_hit_tokens=facts_snapshot.total_cache_hit_tokens,
        )
    )
    persist_token_budget_stats_json(resume_settings().checkpoint_dir, repo_exec=repo_exec)
    return 0


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