tests/test_cli_smoke.py

from __future__ import annotations

from collections.abc import Callable
import io
import importlib
import os
import subprocess
import sys
from pathlib import Path
from types import SimpleNamespace
from typing import NoReturn, cast

import apprc as rc
import pytest
from rich.console import Console
from typer.testing import CliRunner

import kneiff.cli.comfy as cli_comfy
import kneiff.cli.dataset as cli_dataset
from kneiff.config import (
    COMFY_I2I_LORA_1_ENV_KEY,
    COMFY_I2I_LORA_2_ENV_KEY,
    COMFY_I2I_LORA_ENV_KEY,
    COMFY_I2I_LORA_STRENGTH_1_ENV_KEY,
    COMFY_I2I_LORA_STRENGTH_2_ENV_KEY,
    COMFY_I2I_LORA_STRENGTH_ENV_KEY,
    COMFY_I2I_MODEL_DUO_ENV_KEY,
    COMFY_I2I_MODEL_SOLO_ENV_KEY,
    COMFY_LORAS_DIR_1_ENV_KEY,
    COMFY_LORAS_DIR_2_ENV_KEY,
    COMFY_MODELS_DIR_ENV_KEY,
    COMFY_OUTPUT_DIR_ENV_KEY,
    COMFY_POLL_INTERVAL_SECONDS_ENV_KEY,
    COMFY_PROMPT_TIMEOUT_SECONDS_ENV_KEY,
    COMFY_UPSCALE_MODEL_ENV_KEY,
    COMFY_UPSCALE_DENOISE_BASE_ENV_KEY,
    COMFY_UPSCALE_LORA_ENV_KEY,
    COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY,
    COMFY_UPSCALE_LORA_TOKEN_ENV_KEY,
    COMFY_UPSCALE_REFINER_CLIP_ENV_KEY,
    COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY,
    COMFY_UPSCALE_REFINER_UNET_ENV_KEY,
    COMFY_UPSCALE_REFINER_VAE_ENV_KEY,
    COMFY_URL_ENV_KEY,
    COMFY_T2I_LORA_DUO_1_ENV_KEY,
    COMFY_T2I_LORA_DUO_2_ENV_KEY,
    COMFY_T2I_LORA_SOLO_ENV_KEY,
    COMFY_T2I_LORA_STRENGTH_DUO_1_ENV_KEY,
    COMFY_T2I_LORA_STRENGTH_DUO_2_ENV_KEY,
    COMFY_T2I_LORA_STRENGTH_SOLO_ENV_KEY,
    COMFY_T2I_MODEL_DUO_ENV_KEY,
    COMFY_T2I_MODEL_SOLO_ENV_KEY,
    DEFAULT_COMFY_UPSCALE_MODEL,
    DEFAULT_COMFY_UPSCALE_DENOISE_BASE,
    DEFAULT_COMFY_UPSCALE_LORA,
    DEFAULT_COMFY_UPSCALE_LORA_STRENGTH,
    DEFAULT_COMFY_UPSCALE_LORA_TOKEN,
    DEFAULT_COMFY_UPSCALE_REFINER_CLIP,
    DEFAULT_COMFY_UPSCALE_REFINER_CLIP_TYPE,
    DEFAULT_COMFY_UPSCALE_REFINER_UNET,
    DEFAULT_COMFY_UPSCALE_REFINER_VAE,
    KNF_APPRC_TOML_ENV_KEY,
    KNEIFF_RC,
    ComfyConfig,
    KneiffConfig,
    LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_CLIP,
    LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_CLIP_TYPE,
    LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_UNET,
    LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_VAE,
)
from kneiff.infer.comfy.client import ComfyUploadedImage
from kneiff.progress import ProgressUpdate
import kneiff.utils.image.tag_jtp3 as tag_jtp3
from tests._cli_helpers import invoke_cli

pytestmark = pytest.mark.usefixtures("isolated_cli_project")

SHOWCASE_PROMPTS_PATH = (
    Path(__file__).parent / "fixtures" / "projects" / "aurora" / "prompts.knf.yaml"
)


def _write_noob_showcase_prompts(path: Path) -> None:
    """Write one standalone catalog usable by NOOB CLI routing tests."""
    path.write_text(
        """
version: 1
defaults:
  seed: 123
prompts:
  - id: aurora_portrait
    uses: [showcase, training_validation]
    captions:
      nlg:
        - Aurora_Fox portrait
      pony:
        - score_9, Aurora_Fox, portrait
      noob:
        - Aurora_Fox, portrait
""",
        encoding="utf-8",
    )


_DEFAULT_COMFY_INPUT_OPTIONS = {
    ("UpscaleModelLoader", "model_name"): (
        DEFAULT_COMFY_UPSCALE_MODEL,
        "local-upscaler.safetensors",
        "cli-upscaler.safetensors",
        "4x-test.safetensors",
    ),
    ("UNETLoader", "unet_name"): (
        DEFAULT_COMFY_UPSCALE_REFINER_UNET,
        LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_UNET,
        "z-image.safetensors",
    ),
    ("CLIPLoader", "clip_name"): (
        DEFAULT_COMFY_UPSCALE_REFINER_CLIP,
        LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_CLIP,
        "qwen.safetensors",
    ),
    ("CLIPLoader", "type"): (
        DEFAULT_COMFY_UPSCALE_REFINER_CLIP_TYPE,
        LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_CLIP_TYPE,
    ),
    ("VAELoader", "vae_name"): (
        DEFAULT_COMFY_UPSCALE_REFINER_VAE,
        LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_VAE,
    ),
}


class _FakeComfyModelClient:
    def __init__(
        self,
        server_url: str,
        options: dict[tuple[str, str], tuple[str, ...]],
    ) -> None:
        self.server_url = server_url.rstrip("/")
        self.options = options
        self.input_option_requests: list[tuple[str, str]] = []

    def input_options(self, node_class: str, input_name: str) -> tuple[str, ...]:
        key = (node_class, input_name)
        self.input_option_requests.append(key)
        return self.options.get(key, ())


class _CapturedShowcaseProgress:
    """Record aggregate CLI progress without rendering a terminal bar."""

    def __init__(
        self,
        title: str,
        total: int,
        console: Console | None,
    ) -> None:
        self.title = title
        self.total = total
        self.console = console
        self.advances: list[str | None] = []
        self.exit_error: type[BaseException] | None = None

    def __enter__(self) -> _CapturedShowcaseProgress:
        return self

    def __exit__(self, *exc_info: object) -> None:
        error_type = exc_info[0]
        self.exit_error = (
            error_type
            if isinstance(error_type, type) and issubclass(error_type, BaseException)
            else None
        )

    def advance(self, text: str | None = None, *, amount: int = 1) -> None:
        self.advances.extend([text] * amount)

    def step(self, text: str, *, amount: int = 1) -> None:
        self.total += amount
        self.advance(text, amount=amount)

    def apply(self, update: ProgressUpdate) -> None:
        """Record one shared progress update."""
        self.total += update.additional_total
        self.advance(update.description, amount=update.advance)


def _capturing_showcase_progress_factory(
    reporters: list[_CapturedShowcaseProgress],
) -> Callable[..., _CapturedShowcaseProgress]:
    def create(
        title: str,
        total: int = 0,
        console: Console | None = None,
    ) -> _CapturedShowcaseProgress:
        reporter = _CapturedShowcaseProgress(title, total, console)
        reporters.append(reporter)
        return reporter

    return create


def _install_fake_comfy_model_client(
    monkeypatch: pytest.MonkeyPatch,
    *,
    options: dict[tuple[str, str], tuple[str, ...]] | None = None,
) -> _FakeComfyModelClient:
    client = _FakeComfyModelClient(
        "http://127.0.0.1:8188",
        options or dict(_DEFAULT_COMFY_INPUT_OPTIONS),
    )
    monkeypatch.setattr(cli_comfy, "ComfyUiClient", lambda server_url: client)
    return client


@pytest.fixture(autouse=True)
def _isolate_comfy_env(monkeypatch: pytest.MonkeyPatch) -> None:
    import kneiff.infer.comfy.showcase_grid as showcase_grid_module

    owner = rc.schema.owner_for(ComfyConfig)
    for field in owner.fields:
        monkeypatch.delenv(owner.env_key(field.name), raising=False)
    monkeypatch.setattr(
        showcase_grid_module,
        "write_showcase_grid",
        lambda rows, output_subfolder, filename, **kwargs: ComfyUploadedImage(
            name=filename,
            subfolder=output_subfolder,
            type="output",
        ),
    )


def test_cli_help_smoke() -> None:
    runner = CliRunner()

    for args in (
        ["--help"],
        ["comfy", "--help"],
        ["comfy", "showcase", "--help"],
        ["comfy", "upscale", "--help"],
        ["comfy", "outpaint", "--help"],
        ["comfy", "t2i", "--help"],
        ["comfy", "t2i", "solo", "--help"],
        ["comfy", "t2i", "duo", "--help"],
        ["config", "--help"],
        ["config", "paths", "--help"],
        ["config", "show", "--help"],
        ["config", "doctor", "--help"],
        ["config", "setup", "--help"],
        ["config", "set", "--help"],
        ["config", "edit", "--help"],
        ["config", "app", "--help"],
        ["config", "app", "init", "--help"],
        ["config", "storage", "--help"],
        ["config", "storage", "add", "--help"],
        ["config", "storage", "list", "--help"],
        ["config", "storage", "remove", "--help"],
        ["img", "--help"],
        ["img", "tag", "--help"],
        ["llm", "--help"],
        ["llm", "prompt", "--help"],
        ["dataset", "--help"],
        ["dataset", "aspects", "--help"],
        ["dataset", "describe", "--help"],
        ["model", "--help"],
        ["model", "convert", "--help"],
        ["train", "--help"],
        ["train", "start", "--help"],
        ["train", "prepare", "--help"],
        ["train", "runs", "--help"],
        ["train", "review", "--help"],
        ["train", "grid", "--help"],
    ):
        result = invoke_cli(runner, args)
        assert result.exit_code == 0


def test_comfy_t2i_help_exposes_preset_specific_options() -> None:
    runner = CliRunner()

    solo = invoke_cli(runner, ["comfy", "t2i", "solo", "--help"])
    duo = invoke_cli(runner, ["comfy", "t2i", "duo", "--help"])

    assert solo.exit_code == 0
    assert "--pipeline" in solo.output
    assert "--activation-token-1" in solo.output
    assert "--t2i-lora-strength-1" in solo.output
    assert "--num-cleanups" in solo.output
    assert "--i2i-model" in solo.output
    assert "--i2i-lora-strength" in solo.output
    assert duo.exit_code == 0
    assert "--activation-token-2" in duo.output
    assert "--num-cleanups" in duo.output
    assert "--i2i-lora-strength-2" in duo.output


def test_comfy_t2i_solo_requires_pipeline_outside_an_interactive_terminal() -> None:
    runner = CliRunner()

    result = invoke_cli(runner, ["comfy", "t2i", "solo", "studio portrait"])

    assert result.exit_code == 2
    assert "--pipeline" in result.output


def test_showcase_workflow_picker_keeps_project_order_without_duplicates() -> None:
    picker_names = cli_comfy._workflow_picker_preset_names(
        ["pony", "project-only", "anima", "pony", "noob", "project-only"]
    )

    assert picker_names == ["pony", "project-only", "anima", "noob"]


def test_cli_root_help_shows_canonical_image_group_only() -> None:
    runner = CliRunner()

    result = invoke_cli(runner, ["--help"])

    assert result.exit_code == 0
    assert "│ img" in result.output
    assert "│ image" not in result.output
    assert "infer" not in result.output


def test_image_group_alias_is_removed() -> None:
    runner = CliRunner()

    result = invoke_cli(runner, ["image", "--help"])

    assert result.exit_code != 0
    assert "No such command 'image'" in result.output


def test_infer_group_is_removed() -> None:
    runner = CliRunner()

    result = invoke_cli(runner, ["infer", "--help"])

    assert result.exit_code != 0
    assert "No such command 'infer'" in result.output


def test_config_uses_native_apprc_index_path(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    monkeypatch.delenv(KNF_APPRC_TOML_ENV_KEY, raising=False)
    monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config-home"))

    expected_path = tmp_path / "config-home" / "knf" / "knf.apprc.toml"
    assert KNEIFF_RC.spec.index_env_key == KNF_APPRC_TOML_ENV_KEY
    assert KNEIFF_RC.spec.default_index_path() == expected_path
    assert KNEIFF_RC.spec.index_path() == expected_path


def test_cli_help_shows_batch_config_and_resume_run_usage() -> None:
    runner = CliRunner()

    start_result = invoke_cli(runner, ["train", "start", "--help"])
    prepare_result = invoke_cli(runner, ["train", "prepare", "--help"])
    dataset_result = invoke_cli(runner, ["dataset", "plan", "--help"])

    assert start_result.exit_code == 0
    assert "[CONFIG]..." in start_result.output
    assert "knf train start CONFIG..." in start_result.output
    assert "knf train start CONFIG... --resume RUN" in start_result.output
    assert "Pass RUN after --resume" in start_result.output
    assert prepare_result.exit_code == 0
    assert "[CONFIG]..." in prepare_result.output
    assert dataset_result.exit_code == 0
    assert "[CONFIG]..." in dataset_result.output


def test_comfy_showcase_help_shows_prompt_field() -> None:
    runner = CliRunner()
    output_option = "--" + "output"

    result = invoke_cli(runner, ["comfy", "showcase", "--help"])

    assert result.exit_code == 0
    assert "--lora" in result.output
    assert "--training" in result.output
    assert "[STEPS]..." in result.output
    assert "--prompts" in result.output
    assert "--prompt-field" in result.output
    assert output_option not in result.output
    assert "auto" in result.output
    assert "packaged preset" in result.output
    assert "noob-nova" in result.output
    assert "showcase_<preset>.workflow.json" in result.output


def test_comfy_upscale_help_describes_directory_inputs() -> None:
    runner = CliRunner()

    result = invoke_cli(runner, ["comfy", "upscale", "--help"])
    normalized_help = " ".join(result.output.replace("│", " ").split())

    assert result.exit_code == 0
    assert "One or more image files or directories to upscale." in normalized_help
    assert "Directory inputs are scanned recursively by default." in normalized_help
    assert "Include images in nested subdirectories" in normalized_help
    assert "--recursive" in result.output
    assert "--no-recursive" in result.output


def test_comfy_outpaint_help_describes_packaged_file_only_workflows() -> None:
    runner = CliRunner()

    result = invoke_cli(runner, ["comfy", "outpaint", "--help"])
    normalized_help = " ".join(result.output.replace("│", " ").split())

    assert result.exit_code == 0
    assert "Omit them to consume newline-delimited SOURCE/... paths from stdin" in (
        normalized_help
    )
    assert "Directory inputs are rejected" in normalized_help
    assert "--workflow" in result.output
    assert "--safe-border" in result.output
    assert "without ComfyUI inference" in normalized_help
    assert "--seed" in result.output


def test_comfy_t2i_solo_help_uses_positional_prompt() -> None:
    runner = CliRunner()

    result = invoke_cli(runner, ["comfy", "t2i", "solo", "--help"])
    normalized_help = " ".join(result.output.replace("│", " ").split())

    assert result.exit_code == 0
    assert "comfy t2i solo [OPTIONS] PROMPT" in normalized_help
    assert "Scene request used to generate reviewed prompts." in normalized_help


def test_model_showcase_command_is_removed() -> None:
    runner = CliRunner()

    result = invoke_cli(runner, ["model", "showcase", "--help"])

    assert result.exit_code != 0


def test_comfy_config_schema_declares_custom_prefix_fields() -> None:
    owner = rc.schema.owner_for(ComfyConfig)
    expected_env_keys = {
        "url": COMFY_URL_ENV_KEY,
        "models_dir": COMFY_MODELS_DIR_ENV_KEY,
        "loras_dir_1": COMFY_LORAS_DIR_1_ENV_KEY,
        "loras_dir_2": COMFY_LORAS_DIR_2_ENV_KEY,
        "output_dir": COMFY_OUTPUT_DIR_ENV_KEY,
        "t2i_model_solo": COMFY_T2I_MODEL_SOLO_ENV_KEY,
        "t2i_lora_solo": COMFY_T2I_LORA_SOLO_ENV_KEY,
        "t2i_lora_strength_solo": COMFY_T2I_LORA_STRENGTH_SOLO_ENV_KEY,
        "i2i_model_solo": COMFY_I2I_MODEL_SOLO_ENV_KEY,
        "i2i_lora": COMFY_I2I_LORA_ENV_KEY,
        "i2i_lora_strength": COMFY_I2I_LORA_STRENGTH_ENV_KEY,
        "t2i_model_duo": COMFY_T2I_MODEL_DUO_ENV_KEY,
        "t2i_lora_duo_1": COMFY_T2I_LORA_DUO_1_ENV_KEY,
        "t2i_lora_duo_2": COMFY_T2I_LORA_DUO_2_ENV_KEY,
        "t2i_lora_strength_duo_1": COMFY_T2I_LORA_STRENGTH_DUO_1_ENV_KEY,
        "t2i_lora_strength_duo_2": COMFY_T2I_LORA_STRENGTH_DUO_2_ENV_KEY,
        "i2i_model_duo": COMFY_I2I_MODEL_DUO_ENV_KEY,
        "i2i_lora_1": COMFY_I2I_LORA_1_ENV_KEY,
        "i2i_lora_2": COMFY_I2I_LORA_2_ENV_KEY,
        "i2i_lora_strength_1": COMFY_I2I_LORA_STRENGTH_1_ENV_KEY,
        "i2i_lora_strength_2": COMFY_I2I_LORA_STRENGTH_2_ENV_KEY,
        "upscale_model": COMFY_UPSCALE_MODEL_ENV_KEY,
        "upscale_refiner_unet": COMFY_UPSCALE_REFINER_UNET_ENV_KEY,
        "upscale_refiner_clip": COMFY_UPSCALE_REFINER_CLIP_ENV_KEY,
        "upscale_refiner_clip_type": COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY,
        "upscale_refiner_vae": COMFY_UPSCALE_REFINER_VAE_ENV_KEY,
        "upscale_lora": COMFY_UPSCALE_LORA_ENV_KEY,
        "upscale_lora_token": COMFY_UPSCALE_LORA_TOKEN_ENV_KEY,
        "upscale_lora_strength": COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY,
        "upscale_denoise_base": COMFY_UPSCALE_DENOISE_BASE_ENV_KEY,
        "prompt_timeout_seconds": COMFY_PROMPT_TIMEOUT_SECONDS_ENV_KEY,
        "poll_interval_seconds": COMFY_POLL_INTERVAL_SECONDS_ENV_KEY,
    }

    assert owner in KNEIFF_RC.spec.owners
    assert {
        field.name: owner.env_key(field.name) for field in owner.fields
    } == expected_env_keys
    for field_name in expected_env_keys:
        assert owner.config_path_text(field_name) == f"comfy.{field_name}"


def test_kneiff_config_uses_comfy_defaults(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    for env_key in (
        COMFY_URL_ENV_KEY,
        COMFY_MODELS_DIR_ENV_KEY,
        COMFY_LORAS_DIR_1_ENV_KEY,
        COMFY_LORAS_DIR_2_ENV_KEY,
        COMFY_OUTPUT_DIR_ENV_KEY,
        COMFY_T2I_MODEL_SOLO_ENV_KEY,
        COMFY_T2I_LORA_SOLO_ENV_KEY,
        COMFY_T2I_LORA_STRENGTH_SOLO_ENV_KEY,
        COMFY_I2I_MODEL_SOLO_ENV_KEY,
        COMFY_I2I_LORA_ENV_KEY,
        COMFY_I2I_LORA_STRENGTH_ENV_KEY,
        COMFY_T2I_MODEL_DUO_ENV_KEY,
        COMFY_T2I_LORA_DUO_1_ENV_KEY,
        COMFY_T2I_LORA_DUO_2_ENV_KEY,
        COMFY_T2I_LORA_STRENGTH_DUO_1_ENV_KEY,
        COMFY_T2I_LORA_STRENGTH_DUO_2_ENV_KEY,
        COMFY_I2I_MODEL_DUO_ENV_KEY,
        COMFY_I2I_LORA_1_ENV_KEY,
        COMFY_I2I_LORA_2_ENV_KEY,
        COMFY_I2I_LORA_STRENGTH_1_ENV_KEY,
        COMFY_I2I_LORA_STRENGTH_2_ENV_KEY,
        COMFY_UPSCALE_MODEL_ENV_KEY,
        COMFY_UPSCALE_REFINER_UNET_ENV_KEY,
        COMFY_UPSCALE_REFINER_CLIP_ENV_KEY,
        COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY,
        COMFY_UPSCALE_REFINER_VAE_ENV_KEY,
        COMFY_UPSCALE_LORA_ENV_KEY,
        COMFY_UPSCALE_LORA_TOKEN_ENV_KEY,
        COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY,
        COMFY_UPSCALE_DENOISE_BASE_ENV_KEY,
        COMFY_PROMPT_TIMEOUT_SECONDS_ENV_KEY,
        COMFY_POLL_INTERVAL_SECONDS_ENV_KEY,
    ):
        monkeypatch.delenv(env_key, raising=False)

    config = KneiffConfig().comfy

    assert config.url == "http://127.0.0.1:8188"
    assert config.models_dir == ""
    assert config.loras_dir_1 == ""
    assert config.loras_dir_2 == ""
    assert config.output_dir == ""
    assert config.t2i_lora_strength_solo == 1.0
    assert config.i2i_lora_strength == 1.0
    assert config.t2i_lora_strength_duo_1 == 1.0
    assert config.t2i_lora_strength_duo_2 == 1.0
    assert config.i2i_lora_strength_1 == 1.0
    assert config.i2i_lora_strength_2 == 1.0
    assert config.upscale_model == DEFAULT_COMFY_UPSCALE_MODEL
    assert config.upscale_refiner_unet == DEFAULT_COMFY_UPSCALE_REFINER_UNET
    assert config.upscale_refiner_clip == DEFAULT_COMFY_UPSCALE_REFINER_CLIP
    assert config.upscale_refiner_clip_type == DEFAULT_COMFY_UPSCALE_REFINER_CLIP_TYPE
    assert config.upscale_refiner_vae == DEFAULT_COMFY_UPSCALE_REFINER_VAE
    assert config.upscale_lora == DEFAULT_COMFY_UPSCALE_LORA
    assert config.upscale_lora_token == DEFAULT_COMFY_UPSCALE_LORA_TOKEN
    assert config.upscale_lora_strength == DEFAULT_COMFY_UPSCALE_LORA_STRENGTH
    assert config.upscale_denoise_base == DEFAULT_COMFY_UPSCALE_DENOISE_BASE
    assert config.prompt_timeout_seconds == 3600.0
    assert config.poll_interval_seconds == 1.0


def test_kneiff_config_uses_comfy_env_values(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv(COMFY_URL_ENV_KEY, "http://comfy.example")
    monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, "/models")
    monkeypatch.setenv(COMFY_LORAS_DIR_1_ENV_KEY, "/models/models/loras/project")
    monkeypatch.setenv(COMFY_UPSCALE_MODEL_ENV_KEY, "4x-test.safetensors")
    monkeypatch.setenv(COMFY_UPSCALE_REFINER_UNET_ENV_KEY, "z-image.safetensors")
    monkeypatch.setenv(COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, "qwen.safetensors")
    monkeypatch.setenv(COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY, "lumina2")
    monkeypatch.setenv(COMFY_UPSCALE_REFINER_VAE_ENV_KEY, "ae.safetensors")
    monkeypatch.setenv(COMFY_UPSCALE_LORA_ENV_KEY, "project/character.safetensors")
    monkeypatch.setenv(COMFY_UPSCALE_LORA_TOKEN_ENV_KEY, "char_token")
    monkeypatch.setenv(COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY, "0.75")
    monkeypatch.setenv(COMFY_UPSCALE_DENOISE_BASE_ENV_KEY, "0.18")
    monkeypatch.setenv(COMFY_PROMPT_TIMEOUT_SECONDS_ENV_KEY, "12.5")
    monkeypatch.setenv(COMFY_POLL_INTERVAL_SECONDS_ENV_KEY, "0.25")

    config = KneiffConfig().comfy

    assert config == ComfyConfig(
        url="http://comfy.example",
        models_dir="/models",
        loras_dir_1="/models/models/loras/project",
        upscale_model="4x-test.safetensors",
        upscale_refiner_unet="z-image.safetensors",
        upscale_refiner_clip="qwen.safetensors",
        upscale_refiner_clip_type="lumina2",
        upscale_refiner_vae="ae.safetensors",
        upscale_lora="project/character.safetensors",
        upscale_lora_token="char_token",
        upscale_lora_strength=0.75,
        upscale_denoise_base=0.18,
        prompt_timeout_seconds=12.5,
        poll_interval_seconds=0.25,
    )


def test_kneiff_config_loads_every_t2i_environment_field(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    values = {
        COMFY_LORAS_DIR_1_ENV_KEY: "primary",
        COMFY_LORAS_DIR_2_ENV_KEY: "partner",
        COMFY_OUTPUT_DIR_ENV_KEY: "/comfy/output",
        COMFY_T2I_MODEL_SOLO_ENV_KEY: "solo-model.safetensors",
        COMFY_T2I_LORA_SOLO_ENV_KEY: "solo-lora.safetensors",
        COMFY_T2I_LORA_STRENGTH_SOLO_ENV_KEY: "0.11",
        COMFY_I2I_MODEL_SOLO_ENV_KEY: "solo-edit.safetensors",
        COMFY_I2I_LORA_ENV_KEY: "solo-edit-lora.safetensors",
        COMFY_I2I_LORA_STRENGTH_ENV_KEY: "0.22",
        COMFY_T2I_MODEL_DUO_ENV_KEY: "duo-model.safetensors",
        COMFY_T2I_LORA_DUO_1_ENV_KEY: "duo-one.safetensors",
        COMFY_T2I_LORA_DUO_2_ENV_KEY: "duo-two.safetensors",
        COMFY_T2I_LORA_STRENGTH_DUO_1_ENV_KEY: "0.33",
        COMFY_T2I_LORA_STRENGTH_DUO_2_ENV_KEY: "0.44",
        COMFY_I2I_MODEL_DUO_ENV_KEY: "duo-edit.safetensors",
        COMFY_I2I_LORA_1_ENV_KEY: "edit-one.safetensors",
        COMFY_I2I_LORA_2_ENV_KEY: "edit-two.safetensors",
        COMFY_I2I_LORA_STRENGTH_1_ENV_KEY: "0.55",
        COMFY_I2I_LORA_STRENGTH_2_ENV_KEY: "0.66",
    }
    for key, value in values.items():
        monkeypatch.setenv(key, value)

    config = KneiffConfig().comfy

    assert config.loras_dir_1 == "primary"
    assert config.loras_dir_2 == "partner"
    assert config.output_dir == "/comfy/output"
    assert config.t2i_model_solo == "solo-model.safetensors"
    assert config.t2i_lora_solo == "solo-lora.safetensors"
    assert config.t2i_lora_strength_solo == 0.11
    assert config.i2i_model_solo == "solo-edit.safetensors"
    assert config.i2i_lora == "solo-edit-lora.safetensors"
    assert config.i2i_lora_strength == 0.22
    assert config.t2i_model_duo == "duo-model.safetensors"
    assert config.t2i_lora_duo_1 == "duo-one.safetensors"
    assert config.t2i_lora_duo_2 == "duo-two.safetensors"
    assert config.t2i_lora_strength_duo_1 == 0.33
    assert config.t2i_lora_strength_duo_2 == 0.44
    assert config.i2i_model_duo == "duo-edit.safetensors"
    assert config.i2i_lora_1 == "edit-one.safetensors"
    assert config.i2i_lora_2 == "edit-two.safetensors"
    assert config.i2i_lora_strength_1 == 0.55
    assert config.i2i_lora_strength_2 == 0.66


def test_removed_comfy_loras_dir_environment_key_is_ignored(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.delenv(COMFY_LORAS_DIR_1_ENV_KEY, raising=False)
    monkeypatch.setenv("COMFY_LORAS_DIR", "legacy-directory")

    assert KneiffConfig().comfy.loras_dir_1 == ""


def test_kneiff_config_rejects_invalid_comfy_float_env(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.setenv(COMFY_PROMPT_TIMEOUT_SECONDS_ENV_KEY, "slow")

    with pytest.raises(ExceptionGroup, match="ComfySettings") as exc_info:
        KneiffConfig()
    assert "prompt_timeout_seconds" in repr(exc_info.value)


def test_cli_config_error_is_concise_and_does_not_expose_secrets(
    tmp_path: Path,
    isolated_cli_storage: Path,
) -> None:
    secret = "CLI_CONFIG_SECRET_SENTINEL_7a91"
    env = os.environ.copy()
    env.update(
        {
            "COMFY_PROMPT_TIMEOUT_SECONDS": "slow",
            "KNF_LMSTUDIO_API_KEY": secret,
            "KNF_STORAGE": str(isolated_cli_storage),
            "XDG_CONFIG_HOME": str(tmp_path / "config-home"),
        }
    )

    result = subprocess.run(
        [
            sys.executable,
            "-m",
            "kneiff.cli.app",
            "--skip-dotenv-layers",
            "dataset",
            "plan",
        ],
        capture_output=True,
        text=True,
        check=False,
        env=env,
    )

    output = result.stdout + result.stderr
    assert result.returncode == 2
    assert "Invalid Kneiff runtime config" in output
    assert "prompt_timeout_seconds" in output
    assert secret not in output
    assert "Traceback" not in output


def test_non_cli_bundle_resolves_named_storage_after_apprc_bootstrap(
    tmp_path: Path,
) -> None:
    config_home = tmp_path / "config-home"
    storage_root = tmp_path / "named-storage"
    storage_root.mkdir()
    (storage_root / ".env.apprc-storage").touch()
    index_path = config_home / "knf" / "knf.apprc.toml"
    rc.storage.register_storage(
        name="demo",
        root=storage_root,
        path=index_path,
        storage_env_filename=KNEIFF_RC.spec.storage_env_filename,
    )
    env = os.environ.copy()
    env.update(
        {
            "KNF_STORAGE": "demo",
            "XDG_CONFIG_HOME": str(config_home),
        }
    )

    result = subprocess.run(
        [
            sys.executable,
            "-c",
            (
                "from kneiff.config import KNEIFF_RC, KneiffConfig; "
                "KNEIFF_RC.bootstrap(); "
                "print(KneiffConfig().storage.root)"
            ),
        ],
        capture_output=True,
        text=True,
        check=False,
        env=env,
    )

    assert result.returncode == 0, result.stdout + result.stderr
    assert result.stdout.strip() == str(storage_root.resolve())


def test_cli_rejects_unknown_log_level() -> None:
    runner = CliRunner()

    result = invoke_cli(
        runner,
        ["--log-level", "verbose-ish", "dataset", "plan"],
    )

    assert result.exit_code == 2
    assert "Unknown logging level: verbose-ish" in result.output


def test_comfy_showcase_requires_storage(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    monkeypatch.delenv("KNF_STORAGE", raising=False)
    monkeypatch.delenv(KNF_APPRC_TOML_ENV_KEY, raising=False)
    monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config-home"))
    prompts_path = tmp_path / "prompts.yaml"
    prompts_path.write_text(
        """
version: 1
defaults:
  seed: 123
prompts:
  - id: first
    uses: [showcase]
    captions:
      nlg:
        - prose prompt
      pony:
        - pony tags
""",
        encoding="utf-8",
    )

    runner = CliRunner()
    result = invoke_cli(
        runner,
        [
            "comfy",
            "showcase",
            "-w",
            "pony",
            "-p",
            str(prompts_path),
        ],
    )

    assert result.exit_code != 0
    assert "KNF_STORAGE is required" in result.output


def test_comfy_upscale_uses_storage_local_env(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    isolated_cli_storage: Path,
) -> None:
    import kneiff.infer.comfy.upscale as upscale_module
    from kneiff.infer.comfy.upscale import (
        ComfyUpscaleRequest,
        ComfyUpscaleResult,
    )

    (isolated_cli_storage / ".env.apprc-storage").write_text(
        (
            f'{COMFY_UPSCALE_MODEL_ENV_KEY}="local-upscaler.safetensors"\n'
            f'{COMFY_UPSCALE_LORA_ENV_KEY}="missing.safetensors"\n'
            f'{COMFY_UPSCALE_LORA_TOKEN_ENV_KEY}="env_token"\n'
        ),
        encoding="utf-8",
    )
    monkeypatch.delenv(COMFY_UPSCALE_MODEL_ENV_KEY, raising=False)
    image_path = tmp_path / "image.png"
    image_path.write_bytes(b"image")
    requests: list[ComfyUpscaleRequest] = []

    def fake_run_upscale(
        request: ComfyUpscaleRequest,
        **kwargs: object,
    ) -> tuple[ComfyUpscaleResult, ...]:
        requests.append(request)
        return ()

    monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale)
    _install_fake_comfy_model_client(monkeypatch)

    runner = CliRunner()
    result = invoke_cli(runner, ["comfy", "upscale", str(image_path), "--fast"])

    assert result.exit_code == 0, result.output
    assert requests[0].upscale_model == "local-upscaler.safetensors"
    assert requests[0].fast is True
    assert requests[0].lora is None


def test_comfy_upscale_uses_one_aggregate_prompt_progress_task(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    import kneiff.infer.comfy.upscale as upscale_module
    from kneiff.infer.comfy.upscale import (
        ComfyUpscaleRequest,
        ComfyUpscaleResult,
        ProgressCallback as ResultCallback,
    )
    from kneiff.progress import ProgressCallback as WorkProgressCallback

    first = tmp_path / "first.png"
    second = tmp_path / "second.png"
    first.write_bytes(b"image")
    second.write_bytes(b"image")
    reporters: list[_CapturedShowcaseProgress] = []

    def fake_run_upscale(
        request: ComfyUpscaleRequest,
        *,
        progress: ResultCallback | None = None,
        work_progress: WorkProgressCallback | None = None,
        **kwargs: object,
    ) -> tuple[ComfyUpscaleResult, ...]:
        del kwargs
        results: list[ComfyUpscaleResult] = []
        for index, path in enumerate(request.inputs, start=1):
            result = ComfyUpscaleResult(
                input_path=path,
                prompt_id=f"prompt-{index}",
                filename_prefix=f"output-{index}",
                history={},
            )
            results.append(result)
            if work_progress is not None:
                work_progress(
                    ProgressUpdate(
                        f"Image {index}/2 · {path.name}",
                        advance=1,
                    )
                )
            if progress is not None:
                progress(result)
        return tuple(results)

    monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale)
    monkeypatch.setattr(
        cli_comfy,
        "CliProgress",
        _capturing_showcase_progress_factory(reporters),
    )
    _install_fake_comfy_model_client(monkeypatch)

    result = invoke_cli(
        CliRunner(),
        [
            "comfy",
            "upscale",
            str(first),
            str(second),
            "--fast",
            "--model",
            "cli-upscaler.safetensors",
        ],
    )

    assert result.exit_code == 0, result.output
    assert len(reporters) == 1
    assert reporters[0].title == "ComfyUI upscale"
    assert reporters[0].total == 2
    assert reporters[0].advances == [
        "Image 1/2 · first.png",
        "Image 2/2 · second.png",
    ]
    assert f"{first}: output-1" in result.output
    assert f"{second}: output-2" in result.output


def test_dataset_aspects_streams_pruned_paths_with_dynamic_progress(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    import kneiff.training.lora.simpletuner_resolution_report as aspect_report

    reporters: list[_CapturedShowcaseProgress] = []
    progress_descriptions = [
        "Validated dataset paths",
        "Checking manifest image paths 1/2",
        "Built 2 planned geometry variants",
        "Read source dimensions 1/1",
        "Built SimpleTuner artifacts",
        "Analyzing backend fullbody 1/1",
    ]

    def fake_dataset_plan(*_args: object) -> cli_dataset.DatasetSyncPlan:
        return cast(
            cli_dataset.DatasetSyncPlan,
            SimpleNamespace(dataset_name="ready"),
        )

    def fake_build_aspects_analysis(
        plan: cli_dataset.DatasetSyncPlan,
        *,
        process_count: int,
        progress: _CapturedShowcaseProgress,
    ) -> aspect_report.SimpleTunerAspectAnalysis:
        assert plan.dataset_name == "ready"
        assert process_count == 2
        progress.step("Validated dataset paths")
        progress.apply(
            ProgressUpdate(
                "Checking manifest image paths 1/2",
                advance=1,
                additional_total=2,
            )
        )
        progress.apply(
            ProgressUpdate(
                "Built 2 planned geometry variants",
                advance=1,
                additional_total=1,
            )
        )
        progress.apply(
            ProgressUpdate(
                "Read source dimensions 1/1",
                advance=1,
                additional_total=1,
            )
        )
        progress.step("Built SimpleTuner artifacts")
        progress.apply(
            ProgressUpdate(
                "Analyzing backend fullbody 1/1",
                advance=1,
                additional_total=1,
            )
        )
        return cast(aspect_report.SimpleTunerAspectAnalysis, object())

    monkeypatch.setattr(cli_dataset, "_dataset_plan_from_cli", fake_dataset_plan)
    monkeypatch.setattr(
        cli_dataset,
        "_build_dataset_aspects_analysis",
        fake_build_aspects_analysis,
    )
    monkeypatch.setattr(
        aspect_report,
        "format_simpletuner_aspect_analysis",
        lambda _analysis: "Aspect report",
    )
    monkeypatch.setattr(
        aspect_report,
        "pruned_aspect_bucket_source_paths",
        lambda _analysis: ("SOURCE/0-FULLBODY/scene.png",),
    )
    monkeypatch.setattr(
        cli_dataset,
        "CliProgress",
        _capturing_showcase_progress_factory(reporters),
    )

    result = invoke_cli(
        CliRunner(),
        ["dataset", "aspects", "ready", "--processes", "2"],
    )

    assert result.exit_code == 0, result.output
    assert result.stdout == "SOURCE/0-FULLBODY/scene.png\n"
    assert "Aspect report" in result.stderr
    assert len(reporters) == 1
    assert reporters[0].title == "dataset aspects: ready"
    assert reporters[0].total == 7
    assert reporters[0].advances == progress_descriptions
    assert reporters[0].exit_error is None


def test_comfy_outpaint_consumes_dataset_aspect_source_stream(
    isolated_cli_project: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from PIL import Image

    import kneiff.infer.comfy.outpaint as outpaint_module

    source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png"
    source_path.parent.mkdir(parents=True, exist_ok=True)
    Image.new("RGB", (640, 960), color="white").save(source_path)
    requests: list[outpaint_module.ComfyOutpaintRequest] = []

    def fake_run_outpaint(
        request: outpaint_module.ComfyOutpaintRequest,
        **_kwargs: object,
    ) -> tuple[outpaint_module.ComfyOutpaintResult, ...]:
        requests.append(request)
        return ()

    monkeypatch.setattr(outpaint_module, "run_outpaint", fake_run_outpaint)

    result = invoke_cli(
        CliRunner(),
        ["comfy", "outpaint", "-w", "willy"],
        input="SOURCE/0-FULLBODY/scene.png\n",
    )

    assert result.exit_code == 0, result.output
    assert requests[0].inputs == (source_path.resolve(),)
    assert requests[0].square_fraction == 0.25


def test_comfy_outpaint_safe_border_writes_pipeable_local_outputs_without_comfyui(
    isolated_cli_project: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from PIL import Image

    source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png"
    source_path.parent.mkdir(parents=True, exist_ok=True)
    Image.new("RGB", (601, 800), color="white").save(source_path)
    output_root = isolated_cli_project / "safe-border-output"
    monkeypatch.setenv(COMFY_OUTPUT_DIR_ENV_KEY, str(output_root))

    def fail_comfy_client(*_args: object, **_kwargs: object) -> NoReturn:
        raise AssertionError("--safe-border must not construct a ComfyUI client")

    monkeypatch.setattr(cli_comfy, "ComfyUiClient", fail_comfy_client)

    result = invoke_cli(
        CliRunner(),
        ["comfy", "outpaint", "--safe-border"],
        input="SOURCE/0-FULLBODY/scene.png\n",
    )

    assert result.exit_code == 0, result.output
    written_paths = tuple(Path(line) for line in result.stdout.splitlines())
    assert len(written_paths) == 1
    assert written_paths[0].parent.parent == output_root
    assert written_paths[0].parent.name.count("-") == 2
    assert "-safe-border-1" in written_paths[0].name
    assert written_paths[0].is_file()
    assert "Wrote 1 safe-border result" in result.stderr

    explicit_result = invoke_cli(
        CliRunner(),
        ["comfy", "outpaint", str(source_path), "--safe-border"],
        input="",
    )

    assert explicit_result.exit_code == 0, explicit_result.output
    explicit_path = Path(explicit_result.stdout.strip())
    assert explicit_path.is_file()
    assert explicit_path.parent.parent == output_root


def test_comfy_outpaint_safe_border_from_pruned_aspects_needs_no_workflow(
    isolated_cli_project: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from PIL import Image

    import kneiff.datasets.export.aspect_analysis as aspect_analysis
    import kneiff.infer.comfy.outpaint as outpaint_module
    import kneiff.training.lora.simpletuner_resolution_report as aspect_report

    source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png"
    source_path.parent.mkdir(parents=True, exist_ok=True)
    Image.new("RGB", (640, 960), color="white").save(source_path)
    config_path = isolated_cli_project / "configs" / "ANIMA.knf.yaml"
    config_path.parent.mkdir(parents=True, exist_ok=True)
    config_path.write_text(
        'mappings:\n  fullbody:\n    - "0-FULLBODY"\n',
        encoding="utf-8",
    )
    output_root = isolated_cli_project / "safe-border-output"
    monkeypatch.setenv(COMFY_OUTPUT_DIR_ENV_KEY, str(output_root))
    requests: list[outpaint_module.SafeBorderRequest] = []

    def fake_build_aspect_analysis(
        plan: object,
        *,
        process_count: int,
        progress_callback: object,
    ) -> object:
        assert getattr(plan, "dataset_name") == "ANIMA"
        assert process_count == 1
        assert callable(progress_callback)
        return object()

    def fake_run_safe_border(
        request: outpaint_module.SafeBorderRequest,
        **_kwargs: object,
    ) -> tuple[outpaint_module.SafeBorderResult, ...]:
        requests.append(request)
        return ()

    monkeypatch.setattr(
        aspect_analysis,
        "build_dataset_aspect_analysis",
        fake_build_aspect_analysis,
    )
    monkeypatch.setattr(
        aspect_report,
        "pruned_aspect_bucket_source_paths",
        lambda _analysis: ("SOURCE/0-FULLBODY/scene.png",),
    )
    monkeypatch.setattr(outpaint_module, "run_safe_border", fake_run_safe_border)

    result = invoke_cli(
        CliRunner(),
        [
            "comfy",
            "outpaint",
            "-a",
            "--aspects-config",
            "ANIMA",
            "--safe-border",
            "--square-fraction",
            "0.5",
        ],
        input="",
    )

    assert result.exit_code == 0, result.output
    assert requests[0].inputs == (source_path.resolve(),)
    assert requests[0].output_dir == output_root.resolve()
    assert requests[0].square_fraction == 0.5


@pytest.mark.parametrize(
    "inference_option",
    [
        ("--workflow", "willy"),
        ("--url", "http://comfy.test"),
        ("--seed", "42"),
    ],
)
def test_comfy_outpaint_safe_border_rejects_inference_only_options(
    isolated_cli_project: Path,
    inference_option: tuple[str, str],
) -> None:
    from PIL import Image

    source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png"
    source_path.parent.mkdir(parents=True, exist_ok=True)
    Image.new("RGB", (640, 960), color="white").save(source_path)

    result = invoke_cli(
        CliRunner(),
        [
            "comfy",
            "outpaint",
            str(source_path),
            "--safe-border",
            *inference_option,
        ],
        input="",
    )

    assert result.exit_code != 0
    normalized_output = " ".join(result.output.replace("│", " ").split())
    assert "--safe-border cannot be combined" in normalized_output


def test_comfy_outpaint_safe_border_requires_configured_local_output_root(
    isolated_cli_project: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from PIL import Image

    source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png"
    source_path.parent.mkdir(parents=True, exist_ok=True)
    Image.new("RGB", (640, 960), color="white").save(source_path)
    monkeypatch.setenv(COMFY_OUTPUT_DIR_ENV_KEY, "")

    result = invoke_cli(
        CliRunner(),
        ["comfy", "outpaint", str(source_path), "--safe-border"],
        input="",
    )

    assert result.exit_code != 0
    assert f"Set {COMFY_OUTPUT_DIR_ENV_KEY} to use --safe-border" in result.output


def test_comfy_outpaint_interrupt_exits_after_requesting_cancellation(
    isolated_cli_project: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from PIL import Image

    import kneiff.infer.comfy.outpaint as outpaint_module

    source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png"
    source_path.parent.mkdir(parents=True, exist_ok=True)
    Image.new("RGB", (640, 960), color="white").save(source_path)

    def interrupted_run_outpaint(_request: object, **_kwargs: object) -> None:
        raise KeyboardInterrupt

    monkeypatch.setattr(outpaint_module, "run_outpaint", interrupted_run_outpaint)

    result = invoke_cli(
        CliRunner(),
        ["comfy", "outpaint", str(source_path), "-w", "willy"],
        input="",
    )

    assert result.exit_code == 130, result.output
    assert "requested cancellation of its ComfyUI jobs" in result.stderr


def test_comfy_outpaint_from_pruned_aspects_resolves_and_outpaints_sources(
    isolated_cli_project: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from PIL import Image

    import kneiff.datasets.export.aspect_analysis as aspect_analysis
    import kneiff.infer.comfy.outpaint as outpaint_module
    import kneiff.training.lora.simpletuner_resolution_report as aspect_report

    source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png"
    source_path.parent.mkdir(parents=True, exist_ok=True)
    Image.new("RGB", (640, 960), color="white").save(source_path)
    config_path = isolated_cli_project / "configs" / "ANIMA.knf.yaml"
    config_path.parent.mkdir(parents=True, exist_ok=True)
    config_path.write_text(
        'mappings:\n  fullbody:\n    - "0-FULLBODY"\n',
        encoding="utf-8",
    )
    requests: list[outpaint_module.ComfyOutpaintRequest] = []

    def fake_build_aspect_analysis(
        plan: object,
        *,
        process_count: int,
        progress_callback: object,
    ) -> object:
        assert getattr(plan, "dataset_name") == "ANIMA"
        assert process_count == 1
        assert callable(progress_callback)
        return object()

    def fake_run_outpaint(
        request: outpaint_module.ComfyOutpaintRequest,
        **_kwargs: object,
    ) -> tuple[outpaint_module.ComfyOutpaintResult, ...]:
        requests.append(request)
        return ()

    monkeypatch.setattr(
        aspect_analysis,
        "build_dataset_aspect_analysis",
        fake_build_aspect_analysis,
    )
    monkeypatch.setattr(
        aspect_report,
        "pruned_aspect_bucket_source_paths",
        lambda _analysis: ("SOURCE/0-FULLBODY/scene.png",),
    )
    monkeypatch.setattr(outpaint_module, "run_outpaint", fake_run_outpaint)

    result = invoke_cli(
        CliRunner(),
        [
            "comfy",
            "outpaint",
            "-a",
            "--aspects-config",
            "ANIMA",
            "-w",
            "willy",
            "--square-fraction",
            "0.5",
        ],
        input="",
    )

    assert result.exit_code == 0, result.output
    assert "ANIMA: found 1 physically pruned source image(s)." in result.stderr
    assert requests[0].inputs == (source_path.resolve(),)
    assert requests[0].square_fraction == 0.5


def test_comfy_outpaint_from_pruned_aspects_empty_result_is_a_noop(
    isolated_cli_project: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    import kneiff.datasets.export.aspect_analysis as aspect_analysis
    import kneiff.training.lora.simpletuner_resolution_report as aspect_report

    config_path = isolated_cli_project / "configs" / "ANIMA.knf.yaml"
    config_path.parent.mkdir(parents=True, exist_ok=True)
    config_path.write_text(
        'mappings:\n  fullbody:\n    - "0-FULLBODY"\n',
        encoding="utf-8",
    )

    monkeypatch.setattr(
        aspect_analysis,
        "build_dataset_aspect_analysis",
        lambda *_args, **_kwargs: object(),
    )
    monkeypatch.setattr(
        aspect_report,
        "pruned_aspect_bucket_source_paths",
        lambda _analysis: (),
    )

    result = invoke_cli(
        CliRunner(),
        ["comfy", "outpaint", "-a"],
        input="",
    )

    assert result.exit_code == 0, result.output
    assert "No source images are physically pruned" in result.stderr


def test_comfy_outpaint_rejects_pruned_aspects_with_other_input_modes(
    isolated_cli_project: Path,
) -> None:
    from PIL import Image

    source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png"
    source_path.parent.mkdir(parents=True, exist_ok=True)
    Image.new("RGB", (640, 960), color="white").save(source_path)
    config_path = isolated_cli_project / "configs" / "ANIMA.knf.yaml"
    config_path.parent.mkdir(parents=True, exist_ok=True)
    config_path.write_text(
        'mappings:\n  fullbody:\n    - "0-FULLBODY"\n',
        encoding="utf-8",
    )

    result = invoke_cli(
        CliRunner(),
        ["comfy", "outpaint", "-a", str(source_path), "-w", "willy"],
        input="",
    )

    assert result.exit_code != 0
    normalized_output = " ".join(result.output.replace("│", " ").split())
    assert "cannot be combined with positional image inputs" in normalized_output

    piped_result = invoke_cli(
        CliRunner(),
        ["comfy", "outpaint", "-a", "-w", "willy"],
        input="SOURCE/0-FULLBODY/scene.png\n",
    )

    assert piped_result.exit_code != 0
    normalized_piped_output = " ".join(piped_result.output.replace("│", " ").split())
    assert "cannot be combined with positional image inputs" in normalized_piped_output


def test_comfy_outpaint_empty_source_stream_is_a_successful_noop(
    isolated_cli_project: Path,
) -> None:
    del isolated_cli_project
    result = invoke_cli(
        CliRunner(),
        ["comfy", "outpaint"],
        input="",
    )

    assert result.exit_code == 0, result.output
    assert "No outpaint source images received on stdin" in result.stderr


def test_comfy_outpaint_rejects_positional_images_and_source_stream(
    isolated_cli_project: Path,
) -> None:
    from PIL import Image

    source_path = isolated_cli_project / "SOURCE" / "0-FULLBODY" / "scene.png"
    source_path.parent.mkdir(parents=True, exist_ok=True)
    Image.new("RGB", (640, 960), color="white").save(source_path)

    result = invoke_cli(
        CliRunner(),
        ["comfy", "outpaint", str(source_path), "-w", "willy"],
        input="SOURCE/0-FULLBODY/scene.png\n",
    )

    assert result.exit_code != 0
    normalized_output = " ".join(result.output.replace("│", " ").split())
    assert (
        "either positional image files or piped SOURCE/... paths" in normalized_output
    )


def test_comfy_upscale_requires_storage(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    monkeypatch.delenv("KNF_STORAGE", raising=False)
    monkeypatch.delenv(KNF_APPRC_TOML_ENV_KEY, raising=False)
    monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "config-home"))
    image_path = tmp_path / "image.png"
    image_path.write_bytes(b"image")

    runner = CliRunner()
    result = invoke_cli(
        runner,
        [
            "comfy",
            "upscale",
            str(image_path),
            "--model",
            "cli-upscaler.safetensors",
            "--fast",
        ],
    )

    assert result.exit_code != 0
    assert "KNF_STORAGE is required" in result.output


def test_comfy_upscale_quality_uses_brain_off_defaults(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    import kneiff.infer.comfy.upscale as upscale_module
    from kneiff.infer.comfy.upscale import (
        ComfyUpscaleRequest,
        ComfyUpscaleResult,
    )

    for env_key in (
        COMFY_UPSCALE_MODEL_ENV_KEY,
        COMFY_UPSCALE_REFINER_UNET_ENV_KEY,
        COMFY_UPSCALE_REFINER_CLIP_ENV_KEY,
        COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY,
        COMFY_UPSCALE_REFINER_VAE_ENV_KEY,
        COMFY_UPSCALE_LORA_ENV_KEY,
        COMFY_UPSCALE_LORA_TOKEN_ENV_KEY,
        COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY,
        COMFY_UPSCALE_DENOISE_BASE_ENV_KEY,
    ):
        monkeypatch.delenv(env_key, raising=False)
    image_path = tmp_path / "image.png"
    image_path.write_bytes(b"image")
    requests: list[ComfyUpscaleRequest] = []

    def fake_run_upscale(
        request: ComfyUpscaleRequest,
        **kwargs: object,
    ) -> tuple[ComfyUpscaleResult, ...]:
        requests.append(request)
        return ()

    monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale)
    _install_fake_comfy_model_client(monkeypatch)

    runner = CliRunner()
    result = invoke_cli(
        runner, ["--skip-dotenv-layers", "comfy", "upscale", str(image_path)]
    )

    assert result.exit_code == 0, result.output
    assert requests[0].upscale_model == DEFAULT_COMFY_UPSCALE_MODEL
    assert requests[0].quality_workflow == "krea2"
    assert requests[0].refiner is not None
    assert requests[0].refiner.unet == DEFAULT_COMFY_UPSCALE_REFINER_UNET
    assert requests[0].refiner.clip == DEFAULT_COMFY_UPSCALE_REFINER_CLIP


def test_comfy_upscale_z_image_uses_legacy_defaults(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    import kneiff.infer.comfy.upscale as upscale_module
    from kneiff.infer.comfy.upscale import (
        ComfyUpscaleRequest,
        ComfyUpscaleResult,
    )

    for env_key in (
        COMFY_UPSCALE_MODEL_ENV_KEY,
        COMFY_UPSCALE_REFINER_UNET_ENV_KEY,
        COMFY_UPSCALE_REFINER_CLIP_ENV_KEY,
        COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY,
        COMFY_UPSCALE_REFINER_VAE_ENV_KEY,
        COMFY_UPSCALE_LORA_ENV_KEY,
        COMFY_UPSCALE_LORA_TOKEN_ENV_KEY,
        COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY,
        COMFY_UPSCALE_DENOISE_BASE_ENV_KEY,
    ):
        monkeypatch.delenv(env_key, raising=False)
    image_path = tmp_path / "image.png"
    image_path.write_bytes(b"image")
    requests: list[ComfyUpscaleRequest] = []

    def fake_run_upscale(
        request: ComfyUpscaleRequest,
        **kwargs: object,
    ) -> tuple[ComfyUpscaleResult, ...]:
        requests.append(request)
        return ()

    monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale)
    _install_fake_comfy_model_client(monkeypatch)

    runner = CliRunner()
    result = invoke_cli(
        runner,
        ["--skip-dotenv-layers", "comfy", "upscale", str(image_path), "-z"],
    )

    assert result.exit_code == 0, result.output
    assert requests[0].quality_workflow == "z-image"
    assert requests[0].refiner is not None
    assert requests[0].refiner.unet == LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_UNET
    assert requests[0].refiner.clip == LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_CLIP
    assert requests[0].refiner.clip_type == (
        LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_CLIP_TYPE
    )
    assert requests[0].refiner.vae == LEGACY_COMFY_UPSCALE_Z_IMAGE_REFINER_VAE


def test_comfy_upscale_quality_rejects_unavailable_default_non_interactive(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    for env_key in (
        COMFY_UPSCALE_MODEL_ENV_KEY,
        COMFY_UPSCALE_REFINER_UNET_ENV_KEY,
        COMFY_UPSCALE_REFINER_CLIP_ENV_KEY,
        COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY,
        COMFY_UPSCALE_REFINER_VAE_ENV_KEY,
        COMFY_UPSCALE_LORA_ENV_KEY,
        COMFY_UPSCALE_LORA_TOKEN_ENV_KEY,
        COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY,
        COMFY_UPSCALE_DENOISE_BASE_ENV_KEY,
    ):
        monkeypatch.delenv(env_key, raising=False)
    image_path = tmp_path / "image.png"
    image_path.write_bytes(b"image")
    options = dict(_DEFAULT_COMFY_INPUT_OPTIONS)
    options[("UNETLoader", "unet_name")] = ("other-unet.safetensors",)
    _install_fake_comfy_model_client(monkeypatch, options=options)

    runner = CliRunner()
    result = invoke_cli(
        runner, ["--skip-dotenv-layers", "comfy", "upscale", str(image_path)]
    )

    assert result.exit_code != 0
    assert COMFY_UPSCALE_REFINER_UNET_ENV_KEY in result.output
    assert DEFAULT_COMFY_UPSCALE_REFINER_UNET in result.output
    assert "--fast" in result.output


def test_comfy_upscale_quality_cli_passes_refiner_and_lora(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    import kneiff.infer.comfy.upscale as upscale_module
    from kneiff.infer.comfy.upscale import (
        ComfyUpscaleRequest,
        ComfyUpscaleResult,
    )

    models_dir = tmp_path / "models"
    lora_path = models_dir / "models" / "loras" / "character.safetensors"
    lora_path.parent.mkdir(parents=True)
    lora_path.write_bytes(b"lora")
    env_lora_path = models_dir / "models" / "loras" / "env-character.safetensors"
    env_lora_path.write_bytes(b"lora")
    monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
    monkeypatch.setenv(COMFY_UPSCALE_MODEL_ENV_KEY, "4x-test.safetensors")
    monkeypatch.setenv(COMFY_UPSCALE_REFINER_UNET_ENV_KEY, "z-image.safetensors")
    monkeypatch.setenv(COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, "qwen.safetensors")
    monkeypatch.setenv(COMFY_UPSCALE_LORA_ENV_KEY, "env-character.safetensors")
    monkeypatch.setenv(COMFY_UPSCALE_LORA_TOKEN_ENV_KEY, "env_token")
    monkeypatch.setenv(COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY, "0.9")
    image_path = tmp_path / "image.png"
    image_path.write_bytes(b"image")
    requests: list[ComfyUpscaleRequest] = []

    def fake_run_upscale(
        request: ComfyUpscaleRequest,
        **kwargs: object,
    ) -> tuple[ComfyUpscaleResult, ...]:
        requests.append(request)
        return ()

    monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale)
    _install_fake_comfy_model_client(monkeypatch)

    runner = CliRunner()
    result = invoke_cli(
        runner,
        [
            "comfy",
            "upscale",
            str(image_path),
            "--style",
            "flat",
            "--lora",
            "character.safetensors",
            "--token",
            "char_token",
            "--lora-strength",
            "0.4",
            "--seed",
            "123",
        ],
    )

    assert result.exit_code == 0, result.output
    assert requests[0].fast is False
    assert requests[0].style == "flat"
    assert requests[0].refiner is not None
    assert requests[0].refiner.unet == "z-image.safetensors"
    assert requests[0].lora is not None
    assert requests[0].lora.lora_name == "character.safetensors"
    assert requests[0].lora.activation_token == "char_token"
    assert requests[0].lora.strength == 0.4
    assert requests[0].seed_root == 123


def test_comfy_upscale_quality_uses_configured_lora_path_chain(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    import kneiff.infer.comfy.upscale as upscale_module
    from kneiff.infer.comfy.upscale import (
        ComfyUpscaleRequest,
        ComfyUpscaleResult,
    )

    models_dir = tmp_path / "models"
    lora_path = models_dir / "models" / "loras" / "5th-ZI-1" / "2400.safetensors"
    lora_path.parent.mkdir(parents=True)
    lora_path.write_bytes(b"lora")
    monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
    monkeypatch.setenv(COMFY_LORAS_DIR_1_ENV_KEY, "5th-ZI-1")
    monkeypatch.setenv(COMFY_UPSCALE_LORA_ENV_KEY, "2400.safetensors")
    monkeypatch.setenv(COMFY_UPSCALE_LORA_TOKEN_ENV_KEY, "env_token")
    monkeypatch.setenv(COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY, "0.8")
    monkeypatch.setenv(COMFY_UPSCALE_DENOISE_BASE_ENV_KEY, "0.18")
    monkeypatch.setenv(COMFY_UPSCALE_MODEL_ENV_KEY, "4x-test.safetensors")
    monkeypatch.setenv(COMFY_UPSCALE_REFINER_UNET_ENV_KEY, "z-image.safetensors")
    monkeypatch.setenv(COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, "qwen.safetensors")
    image_path = tmp_path / "image.png"
    image_path.write_bytes(b"image")
    requests: list[ComfyUpscaleRequest] = []

    def fake_run_upscale(
        request: ComfyUpscaleRequest,
        **kwargs: object,
    ) -> tuple[ComfyUpscaleResult, ...]:
        requests.append(request)
        return ()

    monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale)
    _install_fake_comfy_model_client(monkeypatch)

    runner = CliRunner()
    result = invoke_cli(runner, ["comfy", "upscale", str(image_path), "-d", "0.19"])

    assert result.exit_code == 0, result.output
    assert requests[0].lora is not None
    assert requests[0].lora.lora_name == "5th-ZI-1/2400.safetensors"
    assert requests[0].lora.activation_token == "env_token"
    assert requests[0].lora.strength == 0.8
    assert requests[0].denoise_base == 0.19


def test_comfy_upscale_quality_allows_configured_lora_without_token(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    import kneiff.infer.comfy.upscale as upscale_module
    from kneiff.infer.comfy.upscale import (
        ComfyUpscaleRequest,
        ComfyUpscaleResult,
    )

    models_dir = tmp_path / "models"
    lora_path = models_dir / "models" / "loras" / "character.safetensors"
    lora_path.parent.mkdir(parents=True)
    lora_path.write_bytes(b"lora")
    monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
    monkeypatch.setenv(COMFY_UPSCALE_LORA_ENV_KEY, "character.safetensors")
    monkeypatch.delenv(COMFY_UPSCALE_LORA_TOKEN_ENV_KEY, raising=False)
    monkeypatch.setenv(COMFY_UPSCALE_MODEL_ENV_KEY, "4x-test.safetensors")
    monkeypatch.setenv(COMFY_UPSCALE_REFINER_UNET_ENV_KEY, "z-image.safetensors")
    monkeypatch.setenv(COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, "qwen.safetensors")
    image_path = tmp_path / "image.png"
    image_path.write_bytes(b"image")
    requests: list[ComfyUpscaleRequest] = []

    def fake_run_upscale(
        request: ComfyUpscaleRequest,
        **kwargs: object,
    ) -> tuple[ComfyUpscaleResult, ...]:
        requests.append(request)
        return ()

    monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale)
    _install_fake_comfy_model_client(monkeypatch)

    runner = CliRunner()
    result = invoke_cli(
        runner, ["--skip-dotenv-layers", "comfy", "upscale", str(image_path)]
    )

    assert result.exit_code == 0, result.output
    assert requests[0].lora is not None
    assert requests[0].lora.activation_token == ""


def test_comfy_upscale_cli_requires_token_with_lora(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    models_dir = tmp_path / "models"
    lora_path = models_dir / "models" / "loras" / "character.safetensors"
    lora_path.parent.mkdir(parents=True)
    lora_path.write_bytes(b"lora")
    monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
    monkeypatch.setenv(COMFY_UPSCALE_MODEL_ENV_KEY, "4x-test.safetensors")
    monkeypatch.setenv(COMFY_UPSCALE_REFINER_UNET_ENV_KEY, "z-image.safetensors")
    monkeypatch.setenv(COMFY_UPSCALE_REFINER_CLIP_ENV_KEY, "qwen.safetensors")
    image_path = tmp_path / "image.png"
    image_path.write_bytes(b"image")
    _install_fake_comfy_model_client(monkeypatch)

    runner = CliRunner()
    result = invoke_cli(
        runner,
        [
            "comfy",
            "upscale",
            str(image_path),
            "--lora",
            "character.safetensors",
        ],
    )

    assert result.exit_code != 0
    assert "--token" in result.output


def test_comfy_upscale_fast_rejects_explicit_quality_options(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    image_path = tmp_path / "image.png"
    image_path.write_bytes(b"image")
    _install_fake_comfy_model_client(monkeypatch)

    runner = CliRunner()
    result = invoke_cli(
        runner,
        [
            "comfy",
            "upscale",
            str(image_path),
            "--fast",
            "--lora",
            "character.safetensors",
        ],
    )

    assert result.exit_code != 0
    assert "--fast" in result.output
    assert "LoRA options" in result.output


def test_comfy_upscale_fast_rejects_z_image_workflow(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    image_path = tmp_path / "image.png"
    image_path.write_bytes(b"image")

    runner = CliRunner()
    result = invoke_cli(
        runner,
        [
            "comfy",
            "upscale",
            str(image_path),
            "--fast",
            "-z",
        ],
    )

    assert result.exit_code != 0
    assert "--z-image" in result.output
    assert "--fast" in result.output


def test_comfy_upscale_explicit_unavailable_model_is_strict(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    image_path = tmp_path / "image.png"
    image_path.write_bytes(b"image")
    _install_fake_comfy_model_client(monkeypatch)
    monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)

    runner = CliRunner()
    result = invoke_cli(
        runner,
        [
            "comfy",
            "upscale",
            str(image_path),
            "--model",
            "missing-upscaler.safetensors",
            "--fast",
        ],
    )

    assert result.exit_code != 0
    assert "missing-upscaler.safetensors" in result.output
    assert "Requested upscale model" in result.output
    assert "Selected upscale model" not in result.output


def test_comfy_upscale_interactive_model_picker_is_current_run_only(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    import kneiff.infer.comfy.upscale as upscale_module
    from kneiff.infer.comfy.upscale import (
        ComfyUpscaleRequest,
        ComfyUpscaleResult,
    )

    image_path = tmp_path / "image.png"
    image_path.write_bytes(b"image")
    monkeypatch.delenv(COMFY_UPSCALE_MODEL_ENV_KEY, raising=False)
    requests: list[ComfyUpscaleRequest] = []
    options = dict(_DEFAULT_COMFY_INPUT_OPTIONS)
    options[("UpscaleModelLoader", "model_name")] = (
        "first-upscaler.safetensors",
        "chosen-upscaler.safetensors",
    )
    actions = iter(("down", "select"))

    def fake_run_upscale(
        request: ComfyUpscaleRequest,
        **kwargs: object,
    ) -> tuple[ComfyUpscaleResult, ...]:
        requests.append(request)
        return ()

    monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale)
    _install_fake_comfy_model_client(monkeypatch, options=options)
    monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
    monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))

    runner = CliRunner()
    result = invoke_cli(runner, ["comfy", "upscale", str(image_path), "--fast"])

    assert result.exit_code == 0, result.output
    assert requests[0].upscale_model == "chosen-upscaler.safetensors"
    assert KneiffConfig().comfy.upscale_model == DEFAULT_COMFY_UPSCALE_MODEL
    assert "Selected upscale model: chosen-upscaler.safetensors" in result.output


def test_comfy_upscale_interactive_lora_token_prompt(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    import kneiff.infer.comfy.upscale as upscale_module
    from kneiff.infer.comfy.upscale import (
        ComfyUpscaleRequest,
        ComfyUpscaleResult,
    )

    models_dir = tmp_path / "models"
    lora_path = models_dir / "models" / "loras" / "character.safetensors"
    lora_path.parent.mkdir(parents=True)
    lora_path.write_bytes(b"lora")
    monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
    for env_key in (
        COMFY_UPSCALE_MODEL_ENV_KEY,
        COMFY_UPSCALE_REFINER_UNET_ENV_KEY,
        COMFY_UPSCALE_REFINER_CLIP_ENV_KEY,
        COMFY_UPSCALE_REFINER_CLIP_TYPE_ENV_KEY,
        COMFY_UPSCALE_REFINER_VAE_ENV_KEY,
        COMFY_UPSCALE_LORA_ENV_KEY,
        COMFY_UPSCALE_LORA_TOKEN_ENV_KEY,
        COMFY_UPSCALE_LORA_STRENGTH_ENV_KEY,
        COMFY_UPSCALE_DENOISE_BASE_ENV_KEY,
    ):
        monkeypatch.delenv(env_key, raising=False)
    image_path = tmp_path / "image.png"
    image_path.write_bytes(b"image")
    requests: list[ComfyUpscaleRequest] = []

    def fake_run_upscale(
        request: ComfyUpscaleRequest,
        **kwargs: object,
    ) -> tuple[ComfyUpscaleResult, ...]:
        requests.append(request)
        return ()

    monkeypatch.setattr(upscale_module, "run_upscale", fake_run_upscale)
    _install_fake_comfy_model_client(monkeypatch)
    monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)

    runner = CliRunner()
    result = invoke_cli(
        runner,
        [
            "--skip-dotenv-layers",
            "comfy",
            "upscale",
            str(image_path),
            "--lora",
            "character.safetensors",
        ],
        input="char_token\n",
    )

    assert result.exit_code == 0, result.output
    assert requests[0].lora is not None
    assert requests[0].lora.activation_token == "char_token"


def test_comfy_showcase_cli_uses_preset_prompt_field(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    import kneiff.infer.comfy.showcase as showcase_module
    from kneiff.infer.comfy.showcase import (
        ShowcaseRunRequest,
        ShowcaseRunResult,
    )

    prompts_path = tmp_path / "prompts.yaml"
    prompts_path.write_text(
        """
version: 1
defaults:
  seed: 123
prompts:
  - id: first
    uses: [showcase]
    captions:
      nlg:
        - prose prompt
      pony:
        - pony tags
      noob:
        - noob tags
""",
        encoding="utf-8",
    )
    requests: list[ShowcaseRunRequest] = []

    def fake_run_showcase(
        request: ShowcaseRunRequest,
        **kwargs: object,
    ) -> ShowcaseRunResult:
        requests.append(request)
        return ShowcaseRunResult(results=())

    monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase)

    runner = CliRunner()
    result = invoke_cli(
        runner,
        [
            "comfy",
            "showcase",
            "-w",
            "pony",
            "-p",
            str(prompts_path),
        ],
    )

    assert result.exit_code == 0, result.output
    assert requests[0].workflow_source.name == "showcase_pony.workflow.json"
    assert requests[0].prompt_field == "pony"
    assert requests[0].showcase_model == "pony"
    assert [prompt.id for prompt in requests[0].prompt_catalog.prompts] == ["first"]
    assert result.output.count("Generating 1 showcase image...") == 1
    assert result.output.count("Showcase complete") == 1

    result = invoke_cli(
        runner,
        [
            "comfy",
            "showcase",
            "-w",
            "noob-willy",
            "-p",
            str(prompts_path),
        ],
    )

    assert result.exit_code == 0, result.output
    assert requests[1].workflow_source.name == "showcase_noob-willy.workflow.json"
    assert requests[1].prompt_field == "noob"
    assert requests[1].showcase_model == "noob"


def test_comfy_showcase_progress_stops_incomplete_after_generation_failure(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    import kneiff.infer.comfy.showcase as showcase_module
    from kneiff.infer.comfy.showcase import (
        ShowcasePromptResult,
        ShowcaseRunRequest,
    )
    from kneiff.infer.comfy.showcase_runner import select_showcase_prompts

    prompts_path = tmp_path / "prompts.yaml"
    prompts_path.write_text(
        """
version: 1
defaults:
  seed: 123
prompts:
  - id: first
    uses: [showcase]
    captions:
      nlg:
        - prose prompt
""",
        encoding="utf-8",
    )
    reporters: list[_CapturedShowcaseProgress] = []

    def fail_run_showcase(
        request: ShowcaseRunRequest,
        *,
        progress: Callable[[ShowcasePromptResult], None] | None = None,
    ) -> NoReturn:
        prompts, _skipped = select_showcase_prompts(
            request.prompt_catalog,
            request.prompt_field,
        )
        prompt = prompts[0]
        if progress is not None:
            progress(
                ShowcasePromptResult(
                    prompt=prompt,
                    prompt_id="queued-first",
                    filename_prefix="first",
                    files=(),
                )
            )
        raise RuntimeError("generation broke")

    monkeypatch.setattr(showcase_module, "run_showcase", fail_run_showcase)
    monkeypatch.setattr(
        cli_comfy,
        "CliProgress",
        _capturing_showcase_progress_factory(reporters),
    )

    result = invoke_cli(
        CliRunner(),
        ["comfy", "showcase", "--workflow", "anima", "-p", str(prompts_path)],
    )

    assert result.exit_code == 1
    assert reporters[0].total == 1
    assert reporters[0].advances == ["first"]
    assert reporters[0].exit_error is RuntimeError
    assert "Showcase failed: generation broke" in result.output
    assert "Showcase complete" not in result.output


def test_comfy_showcase_cli_infers_willy_noob_lora(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    isolated_cli_project: Path,
) -> None:
    import kneiff.infer.comfy.showcase as showcase_module
    from kneiff.infer.comfy.showcase import (
        ShowcaseRunRequest,
        ShowcaseRunResult,
    )

    models_dir = tmp_path / "comfyui-models"
    lora_path = (
        models_dir
        / "models"
        / "loras"
        / "_ladybird"
        / "Rook_Kaefer-v1_0-NOOB-Willy.comfyui.safetensors"
    )
    lora_path.parent.mkdir(parents=True)
    lora_path.write_bytes(b"lora")
    _write_noob_showcase_prompts(isolated_cli_project / "prompts.knf.yaml")
    monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
    requests: list[ShowcaseRunRequest] = []

    def fake_run_showcase(
        request: ShowcaseRunRequest,
        **kwargs: object,
    ) -> ShowcaseRunResult:
        requests.append(request)
        return ShowcaseRunResult(results=())

    monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase)

    runner = CliRunner()
    result = invoke_cli(
        runner,
        [
            "comfy",
            "showcase",
            "--workflow",
            "auto",
            "--lora",
            "_ladybird/Rook_Kaefer-v1_0-NOOB-Willy.comfyui.safetensors",
        ],
    )

    assert result.exit_code == 0, result.output
    assert requests[0].workflow_source.name == "showcase_noob-willy.workflow.json"
    catalog = requests[0].prompt_catalog
    prompt_ids = [prompt.id for prompt in catalog.prompts]
    assert prompt_ids[0] == "character_reference"
    assert "aurora_portrait" in prompt_ids
    assert catalog.overlay_prompt_count == 1
    assert catalog.resolved_prompt_count == catalog.core_prompt_count + 1
    assert requests[0].lora_name == (
        "_ladybird/Rook_Kaefer-v1_0-NOOB-Willy.comfyui.safetensors"
    )
    assert requests[0].prompt_field == "noob"
    assert requests[0].workflow_name == "noob-willy"
    assert "Prompt sources" in result.output
    assert "Workflow  noob-willy" in result.output
    assert "Outputs" in result.output
    assert "Showcase complete" in result.output


def test_comfy_showcase_cli_prompts_for_workflow_when_omitted(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    import kneiff.infer.comfy.showcase as showcase_module
    from kneiff.infer.comfy.showcase import (
        ShowcaseRunRequest,
        ShowcaseRunResult,
    )

    models_dir = tmp_path / "comfyui-models"
    lora_path = (
        models_dir
        / "models"
        / "loras"
        / "_ladybird"
        / "Rook_Kaefer-v1_0-NOOB-Willy.comfyui.safetensors"
    )
    lora_path.parent.mkdir(parents=True)
    lora_path.write_bytes(b"lora")
    prompts_path = tmp_path / "noob-prompts.yaml"
    _write_noob_showcase_prompts(prompts_path)
    monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
    monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
    actions = iter(("down", "select"))
    monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))
    requests: list[ShowcaseRunRequest] = []

    def fake_run_showcase(
        request: ShowcaseRunRequest,
        **kwargs: object,
    ) -> ShowcaseRunResult:
        requests.append(request)
        return ShowcaseRunResult(results=())

    monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase)

    runner = CliRunner()
    result = invoke_cli(
        runner,
        [
            "comfy",
            "showcase",
            "--lora",
            "_ladybird/Rook_Kaefer-v1_0-NOOB-Willy.comfyui.safetensors",
            "--prompts",
            str(prompts_path),
        ],
    )

    assert result.exit_code == 0, result.output
    assert requests[0].workflow_source.name == "showcase_noob-willy.workflow.json"
    assert requests[0].workflow_name == "noob-willy"
    assert requests[0].prompt_field == "noob"


def test_comfy_showcase_cli_prompts_for_generic_noob_workflow(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    import kneiff.infer.comfy.showcase as showcase_module
    from kneiff.infer.comfy.showcase import (
        ShowcaseRunRequest,
        ShowcaseRunResult,
    )

    models_dir = tmp_path / "comfyui-models"
    lora_path = (
        models_dir
        / "models"
        / "loras"
        / "_ladybird"
        / "Rook_Kaefer-v1_0-NOOB.comfyui.safetensors"
    )
    lora_path.parent.mkdir(parents=True)
    lora_path.write_bytes(b"lora")
    prompts_path = tmp_path / "noob-prompts.yaml"
    _write_noob_showcase_prompts(prompts_path)
    monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
    monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
    actions = iter(("down", "select"))
    monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))
    requests: list[ShowcaseRunRequest] = []

    def fake_run_showcase(
        request: ShowcaseRunRequest,
        **kwargs: object,
    ) -> ShowcaseRunResult:
        requests.append(request)
        return ShowcaseRunResult(results=())

    monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase)

    runner = CliRunner()
    result = invoke_cli(
        runner,
        [
            "comfy",
            "showcase",
            "--workflow",
            "auto",
            "--lora",
            "_ladybird/Rook_Kaefer-v1_0-NOOB.comfyui.safetensors",
            "--prompts",
            str(prompts_path),
        ],
    )

    assert result.exit_code == 0, result.output
    assert requests[0].workflow_source.name == "showcase_noob-chenkin.workflow.json"
    assert requests[0].workflow_name == "noob-chenkin"
    assert "Selected workflow: noob-chenkin" in result.output


def test_comfy_showcase_cli_rejects_generic_noob_auto_non_interactive(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    models_dir = tmp_path / "comfyui-models"
    lora_path = (
        models_dir
        / "models"
        / "loras"
        / "_ladybird"
        / "Rook_Kaefer-v1_0-NOOB.comfyui.safetensors"
    )
    lora_path.parent.mkdir(parents=True)
    lora_path.write_bytes(b"lora")
    monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))

    runner = CliRunner()
    result = invoke_cli(
        runner,
        [
            "comfy",
            "showcase",
            "--workflow",
            "auto",
            "--lora",
            "_ladybird/Rook_Kaefer-v1_0-NOOB.comfyui.safetensors",
            "--prompts",
            str(SHOWCASE_PROMPTS_PATH),
        ],
    )

    assert result.exit_code != 0
    assert "LoRA matches multiple workflows" in result.output
    assert "noob-willy" in result.output
    assert "noob-chenkin" in result.output
    assert "noob-base" in result.output
    assert "noob-nova" in result.output
    assert "noob-scrimblosauce" in result.output
    assert "pass --workflow" in result.output


def test_comfy_showcase_cli_prompts_when_packaged_workflow_is_missing(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    import kneiff.infer.comfy.showcase as showcase_module
    from kneiff.infer.comfy.showcase import (
        ShowcaseRunRequest,
        ShowcaseRunResult,
        ShowcaseWorkflowPreset,
    )
    import kneiff.infer.comfy.workflow_presets as workflow_presets

    prompts_path = tmp_path / "prompts.yaml"
    prompts_path.write_text(
        """
version: 1
defaults:
  seed: 123
prompts:
  - id: first
    uses: [showcase]
    captions:
      nlg:
        - prose prompt
""",
        encoding="utf-8",
    )
    models_dir = tmp_path / "comfyui-models"
    lora_path = models_dir / "models" / "loras" / "_ladybird" / "rook.safetensors"
    lora_path.parent.mkdir(parents=True)
    lora_path.write_bytes(b"lora")
    monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
    monkeypatch.setitem(
        workflow_presets.SHOWCASE_WORKFLOW_PRESETS,
        "pony",
        ShowcaseWorkflowPreset(
            name="pony",
            workflow_filename="missing_pony.workflow.json",
            prompt_field="pony",
        ),
    )
    monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
    monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: "select")
    requests: list[ShowcaseRunRequest] = []

    def fake_run_showcase(
        request: ShowcaseRunRequest,
        **kwargs: object,
    ) -> ShowcaseRunResult:
        requests.append(request)
        return ShowcaseRunResult(results=())

    monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase)

    runner = CliRunner()
    result = invoke_cli(
        runner,
        [
            "comfy",
            "showcase",
            "--workflow",
            "pony",
            "--lora",
            "_ladybird/rook.safetensors",
            "--prompts",
            str(prompts_path),
        ],
    )

    assert result.exit_code == 0, result.output
    assert requests[0].workflow_source.name == "showcase_anima.workflow.json"
    assert requests[0].workflow_name == "anima"
    assert "Selected workflow: anima" in result.output


def test_comfy_showcase_lora_picker_descends_into_directories(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    models_dir = tmp_path / "comfyui-models"
    lora_path = models_dir / "models" / "loras" / "nested" / "rook.safetensors"
    lora_path.parent.mkdir(parents=True)
    lora_path.write_bytes(b"lora")
    actions = iter(("select", "down", "select"))
    output = io.StringIO()

    monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))

    selected = cli_comfy.select_showcase_lora_interactively(
        models_dir,
        console=Console(file=output, force_terminal=False),
    )

    assert selected == "nested/rook.safetensors"
    assert "nested/" in output.getvalue()


def test_comfy_showcase_lora_picker_starts_in_configured_directory(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    models_dir = tmp_path / "comfyui-models"
    start_dir = models_dir / "models" / "loras" / "_ladybird" / "v1_0"
    lora_path = start_dir / "rook.safetensors"
    lora_path.parent.mkdir(parents=True)
    lora_path.write_bytes(b"lora")
    output = io.StringIO()

    monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: "select")

    selected = cli_comfy.select_showcase_lora_interactively(
        models_dir,
        console=Console(file=output, force_terminal=False),
        start_dir=start_dir,
    )

    assert selected == "_ladybird/v1_0/rook.safetensors"
    assert "Directory: _ladybird/v1_0" in output.getvalue()


def test_comfy_showcase_lora_picker_start_dir_env_rejects_missing_directory(
    tmp_path: Path,
) -> None:
    models_dir = tmp_path / "comfyui-models"
    missing_dir = models_dir / "models" / "loras" / "_ladybird" / "v1_0"
    config = ComfyConfig(
        models_dir=str(models_dir),
        loras_dir_1=str(missing_dir),
    )

    with pytest.raises(ValueError, match="COMFY_LORAS_DIR_1 directory not found"):
        cli_comfy._resolve_comfy_loras_start_dir(models_dir, config=config)


def test_comfy_showcase_lora_picker_start_dir_env_rejects_outside_lora_root(
    tmp_path: Path,
) -> None:
    models_dir = tmp_path / "comfyui-models"
    outside_dir = tmp_path / "other-loras"
    outside_dir.mkdir()
    config = ComfyConfig(
        models_dir=str(models_dir),
        loras_dir_1=str(outside_dir),
    )

    with pytest.raises(ValueError, match="COMFY_LORAS_DIR_1 must be below"):
        cli_comfy._resolve_comfy_loras_start_dir(models_dir, config=config)


def test_comfy_showcase_training_picker_uses_step_and_cleans_staged_lora(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    isolated_cli_project: Path,
) -> None:
    import kneiff.infer.comfy.showcase as showcase_module
    from kneiff.infer.comfy.showcase import ShowcaseRunRequest, ShowcaseRunResult

    output_root = (
        isolated_cli_project
        / "TRAINING"
        / "ANIMA_1"
        / "_simpletuner-output"
        / "checkpoint-100"
    )
    output_root.mkdir(parents=True)
    (output_root / "pytorch_lora_weights.safetensors").write_bytes(b"raw")
    preferred_path = output_root / "pytorch_lora_weights.comfyui.safetensors"
    preferred_path.write_bytes(b"comfyui")
    second_step = output_root.parent / "checkpoint-200" / "weights.safetensors"
    second_step.parent.mkdir()
    second_step.write_bytes(b"second-step")
    models_dir = tmp_path / "comfyui-models"
    (models_dir / "models" / "loras").mkdir(parents=True)
    monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
    monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
    actions = iter(("select", "down", "select"))
    monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))
    requests: list[ShowcaseRunRequest] = []
    staged_paths: list[Path] = []

    def fake_run_showcase(
        request: ShowcaseRunRequest,
        **kwargs: object,
    ) -> ShowcaseRunResult:
        requests.append(request)
        assert request.lora_name is not None
        staged_path = models_dir / "models" / "loras" / request.lora_name
        assert staged_path.read_bytes() == b"comfyui"
        staged_paths.append(staged_path)
        return ShowcaseRunResult(results=())

    monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase)

    result = invoke_cli(
        CliRunner(),
        [
            "comfy",
            "showcase",
            "-t",
            "100",
            "200",
            "100",
            "--workflow",
            "anima",
        ],
    )

    assert result.exit_code == 0, result.output
    assert len(requests) == 1
    assert requests[0].lora_name is not None
    assert requests[0].lora_name.endswith(
        "ANIMA_1/_simpletuner-output/checkpoint-100/"
        "pytorch_lora_weights.comfyui.safetensors"
    )
    assert "Selected TRAINING LoRA:" in result.output
    assert not staged_paths[0].exists()
    assert not (models_dir / "models" / "loras" / ".kneiff-training").exists()


def test_comfy_showcase_training_hides_root_exports_without_checkpoint_steps(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    isolated_cli_project: Path,
) -> None:
    import kneiff.infer.comfy.showcase as showcase_module
    from kneiff.infer.comfy.showcase import ShowcaseRunRequest, ShowcaseRunResult

    output_root = isolated_cli_project / "TRAINING" / "ANIMA_1" / "_simpletuner-output"
    output_root.mkdir(parents=True)
    root_lora = output_root / "root.comfyui.safetensors"
    root_lora.write_bytes(b"root")
    models_dir = tmp_path / "comfyui-models"
    (models_dir / "models" / "loras").mkdir(parents=True)
    monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
    monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
    requests: list[ShowcaseRunRequest] = []

    def fake_run_showcase(
        request: ShowcaseRunRequest,
        **kwargs: object,
    ) -> ShowcaseRunResult:
        requests.append(request)
        return ShowcaseRunResult(results=())

    monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase)

    result = invoke_cli(
        CliRunner(),
        ["comfy", "showcase", "-t", "--workflow", "anima"],
    )

    assert result.exit_code != 0
    assert "No checkpoint-<step> LoRA outputs" in result.output
    assert not requests


def test_comfy_showcase_training_runs_every_selected_lora_in_one_grid(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    isolated_cli_project: Path,
) -> None:
    import kneiff.infer.comfy.showcase as showcase_module
    import kneiff.infer.comfy.showcase_grid as showcase_grid_module
    from kneiff.infer.comfy.showcase import (
        ShowcasePromptResult,
        ShowcaseRunRequest,
        ShowcaseRunResult,
    )
    from kneiff.infer.comfy.showcase_grid import ShowcaseGridRow
    from kneiff.infer.comfy.showcase_runner import select_showcase_prompts

    output_root = isolated_cli_project / "TRAINING" / "ANIMA_1" / "_simpletuner-output"
    for step in (100, 200):
        lora_path = (
            output_root
            / f"checkpoint-{step}"
            / "pytorch_lora_weights.comfyui.safetensors"
        )
        lora_path.parent.mkdir(parents=True)
        lora_path.write_bytes(str(step).encode("ascii"))
    models_dir = tmp_path / "comfyui-models"
    (models_dir / "models" / "loras").mkdir(parents=True)
    monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
    monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
    actions = iter(("select", "down", "all", "select"))
    monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))
    requests: list[ShowcaseRunRequest] = []
    staged_paths: list[Path] = []
    captured_grid_rows: tuple[ShowcaseGridRow, ...] = ()
    progress_reporters: list[_CapturedShowcaseProgress] = []

    def fake_run_showcase(
        request: ShowcaseRunRequest,
        *,
        progress: Callable[[ShowcasePromptResult], None] | None = None,
    ) -> ShowcaseRunResult:
        requests.append(request)
        assert request.lora_name is not None
        staged_path = models_dir / "models" / "loras" / request.lora_name
        assert staged_path.is_file()
        staged_paths.append(staged_path)
        prompts, skipped_prompt_ids = select_showcase_prompts(
            request.prompt_catalog,
            request.prompt_field,
        )
        results = tuple(
            ShowcasePromptResult(
                prompt=prompt,
                prompt_id=f"queued-{prompt.id}",
                filename_prefix=prompt.id,
                files=(),
            )
            for prompt in prompts
        )
        if progress is not None:
            for prompt_result in results:
                progress(prompt_result)
        return ShowcaseRunResult(
            results=results,
            skipped_prompt_ids=skipped_prompt_ids,
        )

    def fake_write_showcase_grid(
        rows: tuple[ShowcaseGridRow, ...] | list[ShowcaseGridRow],
        *,
        output_subfolder: str,
        filename: str,
        **kwargs: object,
    ) -> ComfyUploadedImage:
        nonlocal captured_grid_rows
        captured_grid_rows = tuple(rows)
        assert all(path.is_file() for path in staged_paths)
        return ComfyUploadedImage(
            name=filename,
            subfolder=output_subfolder,
            type="output",
        )

    monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase)
    monkeypatch.setattr(
        cli_comfy,
        "CliProgress",
        _capturing_showcase_progress_factory(progress_reporters),
    )
    monkeypatch.setattr(
        showcase_grid_module,
        "write_showcase_grid",
        fake_write_showcase_grid,
    )

    result = invoke_cli(
        CliRunner(),
        ["comfy", "showcase", "-t", "100", "200", "--workflow", "anima"],
    )

    assert result.exit_code == 0, result.output
    assert len(requests) == 2
    assert requests[0].generated_at == requests[1].generated_at
    assert all(request.showcase_model == "anima" for request in requests)
    assert [row.label for row in captured_grid_rows] == [
        "ANIMA_1/_simpletuner-output/checkpoint-100/"
        "pytorch_lora_weights.comfyui.safetensors",
        "ANIMA_1/_simpletuner-output/checkpoint-200/"
        "pytorch_lora_weights.comfyui.safetensors",
    ]
    assert all(row.showcase_model == "anima" for row in captured_grid_rows)
    assert all(not path.exists() for path in staged_paths)
    assert not tuple(
        isolated_cli_project.joinpath("TRAINING").glob("*showcase-grid*.jpg")
    )
    assert len(progress_reporters) == 1
    assert progress_reporters[0].title == "Showcase"
    assert progress_reporters[0].total == sum(
        len(row.result.results) for row in captured_grid_rows
    )
    assert progress_reporters[0].advances
    first_progress_text = progress_reporters[0].advances[0]
    final_progress_text = progress_reporters[0].advances[-1]
    assert first_progress_text is not None
    assert final_progress_text is not None
    assert first_progress_text.startswith("LoRA 1/2 · ")
    assert final_progress_text.startswith("LoRA 2/2 · ")
    assert "Selected TRAINING LoRAs:" in result.output
    assert "kneiff-showcase-grid.jpg" in result.output
    assert "LoRA: " not in result.output


@pytest.mark.parametrize(
    ("args", "expected"),
    [
        (["comfy", "showcase", "100"], "STEPS require --training"),
        (
            ["comfy", "showcase", "-t", "--lora", "character.safetensors"],
            "Do not combine --training",
        ),
    ],
)
def test_comfy_showcase_training_rejects_invalid_option_combinations(
    args: list[str],
    expected: str,
) -> None:
    result = invoke_cli(CliRunner(), args)

    assert result.exit_code != 0
    assert expected in result.output


def test_comfy_showcase_training_requires_interactive_terminal() -> None:
    result = invoke_cli(CliRunner(), ["comfy", "showcase", "-t"])

    assert result.exit_code != 0
    assert "requires an interactive" in result.output
    assert "terminal for LoRA selection" in result.output


@pytest.mark.parametrize(
    ("step", "expected"),
    [
        ("0", "positive"),
        ("200", "Available steps: 100"),
    ],
)
def test_comfy_showcase_training_rejects_invalid_or_missing_steps(
    monkeypatch: pytest.MonkeyPatch,
    isolated_cli_project: Path,
    step: str,
    expected: str,
) -> None:
    checkpoint_root = (
        isolated_cli_project
        / "TRAINING"
        / "ANIMA_1"
        / "_simpletuner-output"
        / "checkpoint-100"
    )
    checkpoint_root.mkdir(parents=True)
    (checkpoint_root / "weights.safetensors").write_bytes(b"lora")
    monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)

    result = invoke_cli(CliRunner(), ["comfy", "showcase", "-t", step])

    assert result.exit_code != 0
    assert expected in result.output


def test_comfy_showcase_training_picker_opens_runs_before_checkpoints(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    from kneiff.training.lora.output_discovery import TrainingLoraCandidate

    first_path = tmp_path / "first.comfyui.safetensors"
    second_path = tmp_path / "second.safetensors"
    first_path.write_bytes(b"first")
    second_path.write_bytes(b"second")
    candidates = (
        TrainingLoraCandidate(
            path=first_path,
            relative_path=Path(
                "first_1/_simpletuner-output/checkpoint-50/first.comfyui.safetensors"
            ),
            checkpoint_step=50,
            comfyui_native=True,
        ),
        TrainingLoraCandidate(
            path=second_path,
            relative_path=Path(
                "second_1/_simpletuner-output/checkpoint-100/second.safetensors"
            ),
            checkpoint_step=100,
            comfyui_native=False,
        ),
    )
    actions = iter(("down", "select", "down", "select"))
    output = io.StringIO()
    monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))

    selected = cli_comfy.select_training_lora_interactively(
        candidates,
        console=Console(file=output, force_terminal=False),
    )

    assert selected == (candidates[1],)
    assert "Select TRAINING run" in output.getvalue()
    assert "first_1/ (1 LoRA(s))" in output.getvalue()
    assert candidates[1].relative_path.as_posix() in output.getvalue()


def test_comfy_showcase_training_picker_can_be_cancelled(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    from kneiff.training.lora.output_discovery import TrainingLoraCandidate

    lora_path = tmp_path / "weights.safetensors"
    lora_path.write_bytes(b"lora")
    candidate = TrainingLoraCandidate(
        path=lora_path,
        relative_path=Path(
            "ready_1/_simpletuner-output/checkpoint-100/weights.safetensors"
        ),
        checkpoint_step=100,
        comfyui_native=False,
    )
    monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: "cancel")

    selected = cli_comfy.select_training_lora_interactively(
        (candidate,),
        console=Console(file=io.StringIO(), force_terminal=False),
    )

    assert selected is None


def test_comfy_showcase_training_picker_selects_multiple_loras(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
) -> None:
    from kneiff.training.lora.output_discovery import TrainingLoraCandidate

    candidates = tuple(
        TrainingLoraCandidate(
            path=tmp_path / f"weights-{step}.safetensors",
            relative_path=Path(
                f"ready_1/_simpletuner-output/checkpoint-{step}/weights.safetensors"
            ),
            checkpoint_step=step,
            comfyui_native=False,
        )
        for step in (100, 200)
    )
    for candidate in candidates:
        candidate.path.write_bytes(b"lora")
    actions = iter(("select", "down", "toggle", "down", "toggle", "select"))
    output = io.StringIO()
    monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))

    selected = cli_comfy.select_training_lora_interactively(
        candidates,
        console=Console(file=output, force_terminal=False),
    )

    assert selected == candidates
    assert "Selected TRAINING LoRAs:" in output.getvalue()
    assert "[x]" in output.getvalue()


def test_comfy_showcase_training_infers_workflow_from_relative_path(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    isolated_cli_project: Path,
) -> None:
    import kneiff.infer.comfy.showcase as showcase_module
    from kneiff.infer.comfy.showcase import ShowcaseRunRequest, ShowcaseRunResult

    lora_path = (
        isolated_cli_project
        / "TRAINING"
        / "ANIMA_1"
        / "_simpletuner-output"
        / "checkpoint-100"
        / "weights.comfyui.safetensors"
    )
    lora_path.parent.mkdir(parents=True)
    lora_path.write_bytes(b"lora")
    models_dir = tmp_path / "comfyui-models"
    (models_dir / "models" / "loras").mkdir(parents=True)
    monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
    monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
    actions = iter(("select", "down", "select"))
    monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))
    requests: list[ShowcaseRunRequest] = []

    def fake_run_showcase(
        request: ShowcaseRunRequest,
        **kwargs: object,
    ) -> ShowcaseRunResult:
        requests.append(request)
        return ShowcaseRunResult(results=())

    monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase)

    result = invoke_cli(CliRunner(), ["comfy", "showcase", "-t", "100"])

    assert result.exit_code == 0, result.output
    assert requests[0].workflow_name == "anima"


def test_comfy_showcase_training_falls_back_to_complete_workflow_picker(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    isolated_cli_project: Path,
) -> None:
    import kneiff.infer.comfy.showcase as showcase_module
    from kneiff.infer.comfy.showcase import ShowcaseRunRequest, ShowcaseRunResult

    lora_path = (
        isolated_cli_project
        / "TRAINING"
        / "mystery_1"
        / "_simpletuner-output"
        / "checkpoint-100"
        / "weights.safetensors"
    )
    lora_path.parent.mkdir(parents=True)
    lora_path.write_bytes(b"lora")
    models_dir = tmp_path / "comfyui-models"
    (models_dir / "models" / "loras").mkdir(parents=True)
    monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
    monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
    actions = iter(("select", "down", "select", "select"))
    monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))
    requests: list[ShowcaseRunRequest] = []

    def fake_run_showcase(
        request: ShowcaseRunRequest,
        **kwargs: object,
    ) -> ShowcaseRunResult:
        requests.append(request)
        return ShowcaseRunResult(results=())

    monkeypatch.setattr(showcase_module, "run_showcase", fake_run_showcase)

    result = invoke_cli(CliRunner(), ["comfy", "showcase", "-t", "100"])

    assert result.exit_code == 0, result.output
    assert requests[0].workflow_name == "anima"
    assert "Selected workflow: anima" in result.output


def test_comfy_showcase_training_explicit_auto_keeps_inference_failure(
    monkeypatch: pytest.MonkeyPatch,
    tmp_path: Path,
    isolated_cli_project: Path,
) -> None:
    lora_path = (
        isolated_cli_project
        / "TRAINING"
        / "mystery_1"
        / "_simpletuner-output"
        / "checkpoint-100"
        / "weights.safetensors"
    )
    lora_path.parent.mkdir(parents=True)
    lora_path.write_bytes(b"lora")
    models_dir = tmp_path / "comfyui-models"
    lora_root = models_dir / "models" / "loras"
    lora_root.mkdir(parents=True)
    monkeypatch.setenv(COMFY_MODELS_DIR_ENV_KEY, str(models_dir))
    monkeypatch.setattr(cli_comfy, "interactive_available", lambda console: True)
    actions = iter(("select", "down", "select"))
    monkeypatch.setattr(cli_comfy, "read_picker_action", lambda: next(actions))

    result = invoke_cli(
        CliRunner(),
        ["comfy", "showcase", "-t", "100", "--workflow", "auto"],
    )

    assert result.exit_code != 0
    assert "Could not infer a workflow" in result.output
    assert not (lora_root / ".kneiff-training").exists()


def test_train_help_skips_storage_bootstrap(monkeypatch: pytest.MonkeyPatch) -> None:
    runner = CliRunner()
    monkeypatch.delenv("KNF_STORAGE", raising=False)

    help_result = invoke_cli(runner, ["train", "--help"])
    train_result = invoke_cli(runner, ["train"])

    assert help_result.exit_code == 0, help_result.output
    assert train_result.exit_code == 2
    assert "Training run lifecycle workflows." in train_result.output
    assert "KNF_STORAGE" not in help_result.output
    assert "KNF_STORAGE" not in train_result.output


def test_config_doctor_output_uses_only_knf_env_names() -> None:
    runner = CliRunner()
    forbidden = ("KNF_STORAGE_ROOT", "KNEIFF_APPRC_TOML", "KNEIFF_CONFIG_FILE")
    result = invoke_cli(runner, ["config", "doctor"])

    assert "KNF_STORAGE" in result.output
    assert "KNF_APPRC_TOML" in result.output
    assert not any(name in result.output for name in forbidden)


def test_cli_app_import_keeps_runtime_backends_lazy() -> None:
    script = (
        "import sys\n"
        "import kneiff.cli.app\n"
        "heavy_modules = {\n"
        "    'huggingface_hub', 'numpy', 'openai', 'openpyxl', 'pandas',\n"
        "    'requests', 'safetensors', 'spandrel', 'torch', 'torchvision',\n"
        "    'transformers',\n"
        "}\n"
        "loaded = sorted(name for name in heavy_modules if name in sys.modules)\n"
        "print('\\n'.join(loaded))\n"
        "raise SystemExit(1 if loaded else 0)\n"
    )

    result = subprocess.run(
        [sys.executable, "-c", script],
        capture_output=True,
        text=True,
        check=False,
    )

    assert result.returncode == 0, result.stdout + result.stderr


def test_lora_command_group_is_removed() -> None:
    runner = CliRunner()

    result = invoke_cli(runner, ["lora", "--help"])

    assert result.exit_code != 0


def test_old_diff_subcommand_is_removed() -> None:
    runner = CliRunner()

    result = invoke_cli(runner, ["diff", "--help"])

    assert result.exit_code != 0


def test_img_tag_help_shows_selected_tag_output_options(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.delenv("KNF_STORAGE", raising=False)
    runner = CliRunner()

    result = invoke_cli(runner, ["img", "tag", "--help"])

    assert result.exit_code == 0
    assert "--txt" in result.output
    assert "--comma" in result.output
    assert "-c" in result.output
    assert "Write .txt sidecars next to images" in result.output
    assert "probability CSV output" in result.output
    assert "Legacy JTP-3 snapshot threshold" in result.output


def test_img_tag_cli_passes_text_output_options(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.delenv("KNF_STORAGE", raising=False)
    runner = CliRunner()
    image_path = tmp_path / "image.png"
    image_path.write_bytes(b"image")
    captured: dict[str, tag_jtp3.Jtp3RunConfig] = {}

    def fake_run_jtp3(
        config: tag_jtp3.Jtp3RunConfig,
        snapshot_config: tag_jtp3.Jtp3SnapshotConfig | None = None,
    ) -> subprocess.CompletedProcess[str]:
        captured["config"] = config
        return subprocess.CompletedProcess([], 0)

    monkeypatch.setattr(tag_jtp3, "run_jtp3", fake_run_jtp3)

    result = invoke_cli(runner, ["img", "tag", str(image_path), "--txt", "--comma"])

    assert result.exit_code == 0
    assert captured["config"].write_txt is True
    assert captured["config"].comma_separated is True

    result = invoke_cli(runner, ["img", "tag", str(image_path), "-c"])

    assert result.exit_code == 0
    assert captured["config"].write_txt is False
    assert captured["config"].comma_separated is True


def test_img_tag_rejects_csv_stdout_with_txt(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.delenv("KNF_STORAGE", raising=False)
    runner = CliRunner()
    image_path = tmp_path / "image.png"
    image_path.write_bytes(b"image")

    result = invoke_cli(
        runner, ["img", "tag", str(image_path), "--csv-stdout", "--txt"]
    )

    assert result.exit_code != 0
    assert "--csv-stdout cannot be combined with --txt or --comma" in result.output


def test_package_import_smoke() -> None:
    for module_name in (
        "kneiff",
        "kneiff.clients",
        "kneiff.datasets",
        "kneiff.datasets.manifest",
        "kneiff.datasets.export",
        "kneiff.training",
        "kneiff.training.lora",
        "kneiff.infer",
        "kneiff.cli",
        "kneiff.cli.app",
        "kneiff.main",
        "kneiff.app.workbench",
        "kneiff.utils",
        "kneiff.utils.image.caption",
        "kneiff_dev",
    ):
        assert importlib.import_module(module_name)


def test_removed_import_paths_are_not_shimmed() -> None:
    for module_name in (
        "kneiff.training.captions",
        "kneiff.training.dataset",
        "kneiff.training.manifest",
        "kneiff.training.augmentations",
        "kneiff.inference",
        "kneiff.cli_dataset",
        "kneiff.cli_dataset_report",
        "kneiff.cli_image",
        "kneiff.cli_lora",
        "kneiff.datasets.manifest_to_captions",
        "kneiff.datasets.captions",
        "kneiff.datasets.augment",
        "kneiff.datasets.manifest.core",
    ):
        with pytest.raises(ModuleNotFoundError):
            importlib.import_module(module_name)