fix(run): resolve the bundled default checkpoint at runtime (BE-2994)

comfy_cli/cmdline.py

@@ -822,7 +822,7 @@ def run(
         # against OUR pinned node ids, so mixing them with a user --workflow —
         # whose node ids are arbitrary — is rejected rather than silently
         # misapplied. `preloaded` is handed straight to run's execute path.
-        preloaded: tuple[dict, str, bool] | None = None
+        preloaded: tuple[dict, str, bool, bool] | None = None
         if prompt is not None or set_overrides:
             if workflow is not None:
                 renderer.error(
@@ -831,14 +831,21 @@ def run(
                     hint="drop --workflow to use the bundled default, or edit the workflow file directly",
                 )
                 raise typer.Exit(code=1)
-            from comfy_cli.cql.default_workflow import PromptInjectionError, build_default_workflow
+            from comfy_cli.cql.default_workflow import (
+                PromptInjectionError,
+                build_default_workflow,
+                overrides_set_checkpoint,
+            )
 
             try:
                 injected = build_default_workflow(prompt=prompt, overrides=set_overrides)
             except PromptInjectionError as e:
                 renderer.error(code=e.code, message=str(e), hint=e.hint)
                 raise typer.Exit(code=1) from e
-            preloaded = (injected, "default_text2img", False)
+            # If the user pinned the checkpoint (--set checkpoint=… / 4.ckpt_name=…),
+            # honor it verbatim: runtime resolution is skipped downstream.
+            checkpoint_user_set = overrides_set_checkpoint(set_overrides, injected)
+            preloaded = (injected, "default_text2img", False, checkpoint_user_set)
         elif workflow is None:
             renderer.error(
                 code="prompt_rejected",

comfy_cli/command/run/__init__.py

@@ -38,6 +38,7 @@
 from comfy_cli.command.run.preflight import _detect_partner_nodes as _detect_partner_nodes
 from comfy_cli.command.run.preflight import _fetch_object_info as _fetch_object_info
 from comfy_cli.command.run.preflight import _preflight_validate as _preflight_validate
+from comfy_cli.command.run.preflight import _resolve_default_checkpoint_or_exit as _resolve_default_checkpoint_or_exit
 from comfy_cli.command.run.preflight import fetch_object_info as fetch_object_info
 from comfy_cli.command.run.watcher import _spawn_watcher as _spawn_watcher
 from comfy_cli.command.run.watcher import _tail_state_file as _tail_state_file
@@ -104,7 +105,7 @@ def execute(
     notify: bool = False,
     api_key: str | None = None,
     print_prompt: bool = False,
-    preloaded: tuple[dict, str, bool] | None = None,
+    preloaded: tuple[dict, str, bool, bool] | None = None,
 ):
     # `0.0.0.0` is a wildcard bind, not a connect address. macOS / Windows
     # clients can't reach it; on Linux it happens to resolve to a loopback.
@@ -123,10 +124,13 @@ def execute(
 
     # `preloaded` short-circuits file loading: an in-memory API-format graph
     # (e.g. the `comfy run --prompt` injected default) is handed straight in as
-    # (workflow_dict, display_name, is_ui). Everything downstream is unchanged.
+    # (workflow_dict, display_name, is_ui, checkpoint_user_set). Everything
+    # downstream is unchanged; `checkpoint_user_set` gates runtime checkpoint
+    # resolution for the bundled default (skip it when the user pinned one).
     if preloaded is not None:
-        raw_workflow, workflow_name, is_ui = preloaded
+        raw_workflow, workflow_name, is_ui, checkpoint_user_set = preloaded
     else:
+        checkpoint_user_set = False
         try:
             raw_workflow, workflow_name, is_ui = _load_workflow_file(workflow)
         except WorkflowLoadError as e:
@@ -192,11 +196,30 @@ def execute(
         # foreach item map to stash on the job state at submit time.
         compose_meta = pop_compose_meta(workflow)
 
+    # Partner-API node preflight (below) and runtime checkpoint resolution both
+    # need the server's object_info. `--print-prompt` is a documented
+    # no-server-hit dry-run, so skip the fetch + resolution there and print the
+    # graph as-is; the real submit flow resolves BEFORE the prompt_preview event
+    # so the streamed audit trail advertises the graph we actually submit.
+    object_info: dict = {}
+    if not print_prompt:
+        object_info = _fetch_object_info(host, port)
+
+        # Runtime checkpoint resolution for the bundled `--prompt` default: swap
+        # the pinned checkpoint for one the local server actually has (or
+        # hard-error if it has none). Guarded to the bundled default graph and
+        # skipped when the user pinned the checkpoint explicitly (honor it; let
+        # preflight reject it).
+        if preloaded is not None and workflow_name == "default_text2img" and not checkpoint_user_set:
+            _resolve_default_checkpoint_or_exit(renderer, workflow, object_info, where="local")
+
     # Stream mode: emit the workflow graph so agents have a complete audit
     # trail of what the CLI is about to submit (no-op otherwise).
     renderer.event("prompt_preview", prompt=workflow)
 
-    # --print-prompt: emit/print the workflow and exit without submitting.
+    # --print-prompt: emit/print the workflow and exit without submitting. No
+    # server hit (documented) — the graph is shown as-is, before any
+    # server-dependent checkpoint resolution.
     if print_prompt:
         if renderer.is_pretty():
             print(json.dumps(workflow, indent=2, ensure_ascii=False))
@@ -208,12 +231,6 @@ def execute(
             )
         return
 
-    # Partner-API node preflight. Reject up-front when the workflow
-    # depends on a partner node (Veo/Kling/BFL/Gemini/…) and we have no
-    # credential to inject. If we DO have a credential, plumb it into
-    # extra_data so the partner node finds it server-side — same shape
-    # the cloud submit path uses.
-    object_info = _fetch_object_info(host, port)
     partner_nodes = _detect_partner_nodes(workflow, object_info)
     extra_data: dict | None = None
     if api_key:
@@ -493,7 +510,7 @@ def execute_cloud(
     timeout: int = 600,
     notify: bool = False,
     print_prompt: bool = False,
-    preloaded: tuple[dict, str, bool] | None = None,
+    preloaded: tuple[dict, str, bool, bool] | None = None,
 ):
     """Run a workflow against Comfy Cloud via the stored OAuth session.
 
@@ -508,8 +525,9 @@ def execute_cloud(
 
     renderer = get_renderer()
     if preloaded is not None:
-        raw_workflow, workflow_name, is_ui = preloaded
+        raw_workflow, workflow_name, is_ui, checkpoint_user_set = preloaded
     else:
+        checkpoint_user_set = False
         try:
             raw_workflow, workflow_name, is_ui = _load_workflow_file(workflow)
         except WorkflowLoadError as e:
@@ -570,6 +588,23 @@ def execute_cloud(
     # its foreach item map to stash on the job state at submit time.
     compose_meta = pop_compose_meta(parsed_workflow)
 
+    # Cloud path uses cached/bundled object_info (no live server needed). Load
+    # it up front so checkpoint resolution can run BEFORE the preview/print
+    # below — the audit trail must advertise the graph we actually submit.
+    try:
+        from comfy_cli.cql.engine import _load_from_target
+
+        cloud_object_info = _load_from_target(mode="cloud")
+    except Exception:  # noqa: BLE001
+        cloud_object_info = {}
+
+    # Runtime checkpoint resolution for the bundled `--prompt` default (mirrors
+    # the local path): swap the pinned checkpoint for one Comfy Cloud actually
+    # has. Guarded to the bundled default and skipped when the user pinned the
+    # checkpoint explicitly. Cloud fails open on an empty enum (per-job models).
+    if preloaded is not None and workflow_name == "default_text2img" and not checkpoint_user_set:
+        _resolve_default_checkpoint_or_exit(renderer, parsed_workflow, cloud_object_info, where="cloud")
+
     if print_prompt:
         # Documented dry-run: show the API-format graph that WOULD be sent and
         # exit WITHOUT POSTing. Mirrors local execute()'s print_prompt branch.
@@ -585,14 +620,6 @@ def execute_cloud(
         raise typer.Exit(code=0)
 
     # Pre-submit validation via pure-Python CQL engine.
-    # Cloud path uses cached/bundled object_info (no live server needed).
-    try:
-        from comfy_cli.cql.engine import _load_from_target
-
-        cloud_object_info = _load_from_target(mode="cloud")
-    except Exception:  # noqa: BLE001
-        cloud_object_info = {}
-
     _preflight_validate(renderer, parsed_workflow, cloud_object_info, target_label="cloud")
 
     target = resolve_target(where="cloud")

comfy_cli/command/run/preflight.py

@@ -111,6 +111,67 @@ def _preflight_validate(renderer, workflow: dict, object_info: dict, *, target_l
             pprint(f"[yellow]⚠ {w.get('field', '?')}: {w.get('message', '')}[/yellow]")
 
 
+def _resolve_default_checkpoint_or_exit(renderer, workflow: dict, object_info: dict, *, where: str) -> None:
+    """Runtime-resolve the bundled default's pinned checkpoint against the
+    target, in place, then report the outcome through the renderer.
+
+    Call ONLY for the bundled default graph (``workflow_name ==
+    "default_text2img"``) when the user did NOT explicitly ``--set`` the
+    checkpoint. Three outcomes:
+
+    - pinned present / can't tell (object_info empty or not enumerated) → no-op
+      (fail open — preflight + the server decide);
+    - pinned absent but the target has ≥1 checkpoint → substitute the first
+      available one and emit a ``checkpoint_substituted`` note;
+    - target positively has zero checkpoints → for ``where="local"`` a hard
+      ``no_checkpoint_available`` error (exit 1) instead of a cryptic
+      server-side reject; for ``where="cloud"`` a no-op (fail open), since Comfy
+      Cloud provisions its models per-job and the cached enum can't prove the
+      run would fail.
+
+    ``where`` is ``"local"`` or ``"cloud"`` and drives the target label + hint.
+    """
+    from comfy_cli.cql.default_workflow import resolve_default_checkpoint
+
+    target_label = "the local server" if where == "local" else "Comfy Cloud"
+    _, res = resolve_default_checkpoint(workflow, object_info, target=target_label)
+
+    # Comfy Cloud provisions its models per-job at runtime, so an empty
+    # checkpoint enum in the cached/bundled cloud object_info does NOT mean the
+    # run would fail — hard-erroring there would wrongly block valid default
+    # cloud submits. Only the local path (where the enum reflects what's
+    # actually installed) treats a positively-empty enum as a hard stop.
+    if res.no_checkpoint and where == "cloud":
+        return
+
+    if res.no_checkpoint:
+        # Only reachable for the local path (cloud returned above).
+        hint = (
+            "download a checkpoint, e.g. `comfy model download --url "
+            "https://huggingface.co/Comfy-Org/stable-diffusion-v1-5-archive/resolve/main/"
+            "v1-5-pruned-emaonly-fp16.safetensors`, then re-run — or `--set checkpoint=<name>`"
+        )
+        renderer.error(
+            code="no_checkpoint_available",
+            message=(
+                f"the bundled default text2img workflow needs a checkpoint, but {target_label} has none installed"
+            ),
+            hint=hint,
+            details={"where": where},
+        )
+        raise typer.Exit(code=1)
+
+    if res.note:
+        # Event fires in NDJSON/stream mode only; the pretty line covers humans.
+        renderer.event("checkpoint_substituted", message=res.note, checkpoint=res.substituted_to, where=where)
+        if renderer.is_pretty():
+            from rich.markup import escape
+
+            # res.note embeds a target-provided checkpoint name; escape it so a
+            # name containing Rich tags can't inject terminal markup.
+            pprint(f"[yellow]⚠ {escape(res.note)}[/yellow]")
+
+
 def _fetch_object_info(host: str, port: int) -> dict:
     """Fetch object_info for partner-node detection + validation. Fail open."""
     try:

comfy_cli/cql/data/default_text2img.json

@@ -18,7 +18,7 @@
   "4": {
     "class_type": "CheckpointLoaderSimple",
     "inputs": {
-      "ckpt_name": "v1-5-pruned-emaonly.ckpt"
+      "ckpt_name": "v1-5-pruned-emaonly-fp16.safetensors"
     },
     "_meta": {"title": "Load Checkpoint"}
   },

comfy_cli/cql/default_workflow.py

@@ -16,8 +16,14 @@
 
 import json
 import math
+from dataclasses import dataclass
 from importlib import resources
 
+# The name pinned in ``data/default_text2img.json`` node "4" ``ckpt_name``.
+# Runtime resolution (``resolve_default_checkpoint``) swaps this for a checkpoint
+# the target actually has when it's absent; keep the two in sync.
+DEFAULT_CHECKPOINT_NAME = "v1-5-pruned-emaonly-fp16.safetensors"
+
 # -- Pinned node ids (must match data/default_text2img.json) --
 CHECKPOINT_LOADER_ID = "4"
 POSITIVE_PROMPT_ID = "6"
@@ -219,3 +225,128 @@ def build_default_workflow(*, prompt: str | None = None, overrides: list[str] |
         _apply_set(workflow, node_id, field, value)
 
     return workflow
+
+
+def overrides_set_checkpoint(overrides: list[str] | None, workflow: dict) -> bool:
+    """True if any ``--set`` override targets the checkpoint (node ``"4"``
+    ``ckpt_name``), via an alias (``checkpoint``/``ckpt``) or the raw
+    ``4.ckpt_name`` form.
+
+    Used by the caller to decide whether the user pinned the checkpoint
+    explicitly — if so, runtime resolution is skipped and the value is honored
+    verbatim. ``workflow`` must be a built default graph so addresses resolve;
+    malformed entries are ignored here (``build_default_workflow`` already
+    validated/rejected them upstream).
+    """
+    for raw in overrides or []:
+        if "=" not in raw:
+            continue
+        address = raw.partition("=")[0].strip()
+        try:
+            node_id, field = _resolve_address(address, workflow)
+        except PromptInjectionError:
+            continue
+        if (node_id, field) == (CHECKPOINT_LOADER_ID, "ckpt_name"):
+            return True
+    return False
+
+
+@dataclass(frozen=True)
+class CheckpointResolution:
+    """Outcome of :func:`resolve_default_checkpoint`.
+
+    - ``note``/``substituted_to`` are set only when the pinned checkpoint was
+      absent and a different one was substituted.
+    - ``no_checkpoint`` is True ONLY when ``object_info`` positively enumerated
+      an EMPTY checkpoint list (the target has zero checkpoints). It stays False
+      when we can't tell (``object_info`` absent/empty, or it didn't enumerate
+      ``CheckpointLoaderSimple.ckpt_name``) so callers fail open there.
+    """
+
+    note: str | None = None
+    substituted_to: str | None = None
+    no_checkpoint: bool = False
+
+
+def _checkpoint_enum(object_info: dict) -> list | None:
+    """Return the ``CheckpointLoaderSimple.ckpt_name`` option list from
+    ``object_info``, or ``None`` when it isn't enumerated at all.
+
+    The raw object_info shape is ``ckpt_name: [[<name>, …], {opts}]`` — the
+    option list is element 0. ``None`` (not an empty list) means "can't tell"
+    so the caller can distinguish a positively-empty enum from an absent one.
+    """
+    if not isinstance(object_info, dict):
+        # A non-object /object_info payload (e.g. a hostile or misbehaving
+        # server returning ``[]``) means "can't tell" — fail open, mirroring
+        # Graph.from_object_info's isinstance guard rather than crashing.
+        return None
+    node = object_info.get("CheckpointLoaderSimple")
+    if not isinstance(node, dict):
+        return None
+    inp = node.get("input")
+    if not isinstance(inp, dict):
+        return None
+    req = inp.get("required")
+    if not isinstance(req, dict):
+        return None
+    spec = req.get("ckpt_name")
+    if isinstance(spec, list) and spec and isinstance(spec[0], list):
+        return spec[0]
+    return None
+
+
+def resolve_default_checkpoint(
+    workflow: dict, object_info: dict, *, target: str = "the server"
+) -> tuple[dict, CheckpointResolution]:
+    """Resolve the bundled default's pinned checkpoint against a live target.
+
+    Pure and offline-testable. Given the bundled default graph and a target's
+    ``object_info``, decide what checkpoint node ``"4"`` should carry:
+
+    - pinned name present in the target's enum → no change;
+    - pinned absent but the enum is non-empty → substitute the first available
+      checkpoint and return a ``note``;
+    - enum positively empty → leave unchanged, flag ``no_checkpoint`` (the
+      caller emits an actionable error);
+    - enum absent / ``object_info`` empty → leave unchanged, no flag (fail open).
+
+    Mutates ``workflow`` in place on substitution and returns it alongside the
+    :class:`CheckpointResolution`. Callers must guard this to the bundled
+    default graph only (``workflow_name == "default_text2img"``).
+    """
+    node = workflow.get(CHECKPOINT_LOADER_ID)
+    inputs = node.get("inputs") if isinstance(node, dict) else None
+    if not isinstance(inputs, dict):
+        return workflow, CheckpointResolution()
+
+    enum = _checkpoint_enum(object_info)
+    if enum is None:
+        # Not enumerated (fresh/unfetched object_info) — fail open.
+        return workflow, CheckpointResolution()
+    if not enum:
+        # Positively empty: the target has zero checkpoints installed.
+        return workflow, CheckpointResolution(no_checkpoint=True)
+
+    pinned = inputs.get("ckpt_name")
+    if pinned in enum:
+        return workflow, CheckpointResolution()
+
+    # ComfyUI enumerates checkpoints by their path relative to the models dir,
+    # so a pinned bare filename won't exact-match the same file living in a
+    # subfolder (e.g. ``SD1.5/v1-5-…safetensors``). Prefer a basename match to
+    # the *intended* checkpoint before falling back to an arbitrary substitute.
+    if isinstance(pinned, str):
+        pinned_base = pinned.rsplit("/", 1)[-1]
+        for entry in enum:
+            if isinstance(entry, str) and entry.rsplit("/", 1)[-1] == pinned_base:
+                inputs["ckpt_name"] = entry
+                return workflow, CheckpointResolution()
+
+    replacement = enum[0]
+    inputs["ckpt_name"] = replacement
+    note = (
+        f"default checkpoint {pinned} not found on {target}; using {replacement} "
+        f"instead (override with --set checkpoint=<name>)"
+    )
+    return workflow, CheckpointResolution(note=note, substituted_to=replacement)