feat(nodes): add object-info snapshots

comfy_cli/command/nodes.py

@@ -20,14 +20,20 @@
 from __future__ import annotations
 
 import difflib
+import json
+import os
+import tempfile
+from pathlib import Path
 from typing import Annotated, Any
 
 import typer
 
 from comfy_cli import tracking
 from comfy_cli.cql.engine import Graph, LoadError
+from comfy_cli.http import authed_urlopen
 from comfy_cli.output import get_renderer, rprint
 from comfy_cli.output.sanitize import sanitize_markup
+from comfy_cli.target import Target, resolve_target
 
 app = typer.Typer(no_args_is_help=True, help="Introspect ComfyUI node classes (inputs, outputs, categories).")
 
@@ -1040,6 +1046,208 @@ def categories_cmd(
     renderer.emit(payload, command="nodes categories")
 
 
+# ---------------------------------------------------------------------------
+# snapshot — persist raw object_info for offline validation
+# ---------------------------------------------------------------------------
+
+
+_OBJECT_INFO_SNAPSHOT_CHUNK_BYTES = 1024 * 1024
+_OBJECT_INFO_SNAPSHOT_MAX_BYTES = 128 * 1024 * 1024
+_OBJECT_INFO_PARSE_CHUNK_CHARS = 1024 * 1024
+_OBJECT_INFO_KEY_MAX_CHARS = 64 * 1024
+_OBJECT_INFO_ENTRY_MAX_CHARS = 8 * 1024 * 1024
+
+
+def _validate_object_info_stream(path: Path) -> int:
+    """Validate a top-level object_info mapping without loading it all at once."""
+    decoder = json.JSONDecoder()
+    with path.open(encoding="utf-8") as handle:
+        buffer = ""
+        pos = 0
+        eof = False
+
+        def refill(*, compact: bool) -> bool:
+            nonlocal buffer, pos, eof
+            if compact and pos:
+                buffer = buffer[pos:]
+                pos = 0
+            chunk = handle.read(_OBJECT_INFO_PARSE_CHUNK_CHARS)
+            if not chunk:
+                eof = True
+                return False
+            buffer += chunk
+            return True
+
+        def skip_whitespace() -> None:
+            nonlocal pos
+            while True:
+                while pos < len(buffer) and buffer[pos] in " \t\r\n":
+                    pos += 1
+                if pos < len(buffer) or eof:
+                    return
+                refill(compact=True)
+
+        def current() -> str | None:
+            skip_whitespace()
+            if pos < len(buffer):
+                return buffer[pos]
+            return None
+
+        def decode_value(label: str, max_chars: int):
+            nonlocal buffer, pos, eof
+            start = pos
+            while True:
+                try:
+                    value, end = decoder.raw_decode(buffer, pos)
+                except json.JSONDecodeError as e:
+                    if len(buffer) - start > max_chars:
+                        raise ValueError(f"object_info {label} exceeds the {max_chars:,}-character limit") from e
+                    chunk = handle.read(_OBJECT_INFO_PARSE_CHUNK_CHARS)
+                    if not chunk:
+                        eof = True
+                        raise ValueError("object_info response is not valid JSON") from e
+                    buffer += chunk
+                    continue
+                if end - start > max_chars:
+                    raise ValueError(f"object_info {label} exceeds the {max_chars:,}-character limit")
+                pos = end
+                return value
+
+        if not refill(compact=False) or current() != "{":
+            raise ValueError("object_info response is valid JSON but not a node catalog")
+        pos += 1
+        classes = 0
+
+        while True:
+            token = current()
+            if token == "}":
+                pos += 1
+                break
+            if token is None:
+                raise ValueError("object_info response is not valid JSON")
+
+            key = decode_value("class name", _OBJECT_INFO_KEY_MAX_CHARS)
+            if not isinstance(key, str):
+                raise ValueError("object_info response is valid JSON but not a node catalog")
+            if current() != ":":
+                raise ValueError("object_info response is not valid JSON")
+            pos += 1
+            skip_whitespace()
+            entry = decode_value("catalog entry", _OBJECT_INFO_ENTRY_MAX_CHARS)
+            if not isinstance(entry, dict) or not ("input" in entry or "category" in entry):
+                raise ValueError("object_info response is valid JSON but not a node catalog")
+            classes += 1
+
+            token = current()
+            if token == ",":
+                pos += 1
+                if current() in (None, "}"):
+                    raise ValueError("object_info response is not valid JSON")
+            elif token == "}":
+                pos += 1
+                break
+            else:
+                raise ValueError("object_info response is not valid JSON")
+
+            if pos > _OBJECT_INFO_PARSE_CHUNK_CHARS:
+                buffer = buffer[pos:]
+                pos = 0
+
+        if classes == 0:
+            raise ValueError("object_info response is valid JSON but not a node catalog")
+        if current() is not None:
+            raise ValueError("object_info response is not valid JSON")
+        return classes
+
+
+def _resolve_snapshot_target(where: str | None, host: str | None, port: int | None) -> Target:
+    from comfy_cli import where as where_module
+
+    decision = where_module.resolve_default_or_exit(flag=where)
+    mode = decision.target.value
+    get_renderer().where = mode
+    if mode == "local":
+        from comfy_cli.host_port import report_usage_error, resolve_host_port
+
+        with report_usage_error(get_renderer()):
+            host, port = resolve_host_port(host, port)
+    return resolve_target(where=mode, host=host, port=port)
+
+
+def _stream_object_info_snapshot(target: Target, output: Path) -> dict[str, int]:
+    output = output.expanduser()
+    fd, temp_name = tempfile.mkstemp(dir=str(output.parent), prefix=f"{output.name}.", suffix=".tmp")
+    temp_path = Path(temp_name)
+    total = 0
+    try:
+        with os.fdopen(fd, "wb") as dst:
+            with authed_urlopen(target.url("object_info"), target, timeout=60.0) as response:
+                while chunk := response.read(_OBJECT_INFO_SNAPSHOT_CHUNK_BYTES):
+                    total += len(chunk)
+                    if total > _OBJECT_INFO_SNAPSHOT_MAX_BYTES:
+                        raise ValueError("object_info response exceeds the 128 MiB snapshot limit")
+                    dst.write(chunk)
+            dst.flush()
+            os.fsync(dst.fileno())
+
+        try:
+            classes = _validate_object_info_stream(temp_path)
+        except (UnicodeDecodeError, RecursionError) as e:
+            raise ValueError("object_info response is not valid JSON") from e
+
+        os.replace(temp_path, output)
+        return {"bytes": total, "classes": classes}
+    finally:
+        temp_path.unlink(missing_ok=True)
+
+
+@app.command(
+    "snapshot",
+    help="Save the target's raw object_info catalog for offline validation.",
+)
+@tracking.track_command("nodes")
+def snapshot_cmd(
+    output: Annotated[
+        Path,
+        typer.Option(
+            "--output",
+            "-o",
+            help="Destination object_info JSON file (written atomically).",
+        ),
+    ],
+    where: Annotated[
+        str | None,
+        typer.Option(
+            "--where",
+            show_default=False,
+            help="'local' or 'cloud'; defaults through the normal routing chain.",
+        ),
+    ] = None,
+    host: Annotated[str | None, typer.Option(show_default=False)] = None,
+    port: Annotated[int | None, typer.Option(show_default=False)] = None,
+):
+    renderer = get_renderer()
+    try:
+        target = _resolve_snapshot_target(where, host, port)
+        result = _stream_object_info_snapshot(target, output)
+    except (OSError, ValueError, RuntimeError, RecursionError) as e:
+        renderer.error(
+            code="nodes_snapshot_failed",
+            message=f"Could not save object_info snapshot: {e}",
+            hint="check the target, destination directory, and available disk space",
+            details={"output": str(output)},
+        )
+        raise typer.Exit(code=1) from e
+
+    payload = {"output": str(output.expanduser()), **result}
+    if renderer.is_pretty():
+        rprint(
+            f"[green]✓[/green] {result['classes']:,} node classes ({result['bytes']:,} bytes) → "
+            f"{sanitize_markup(output.expanduser())}"
+        )
+    renderer.emit(payload, command="nodes snapshot", changed=True)
+
+
 # ---------------------------------------------------------------------------
 # refresh — object_info is fetched live; the annotation data is what's cached
 # ---------------------------------------------------------------------------

comfy_cli/discovery.py

@@ -71,6 +71,7 @@
     "comfy nodes path": "nodes",
     "comfy nodes types": "nodes",
     "comfy nodes categories": "nodes",
+    "comfy nodes snapshot": "nodes",
     "comfy nodes refresh": "nodes",
     # workflow editing
     "comfy workflow slots": "workflow",

comfy_cli/error_codes.py

@@ -602,6 +602,11 @@ class ErrorCode:
         "Surfaced in `data.warnings[]` (not as an error envelope) so the command still succeeds.",
         "re-run once the server/session is reachable to get a fresh schema",
     ),
+    ErrorCode(
+        "nodes_snapshot_failed",
+        "The target object_info catalog could not be streamed, validated, or written atomically.",
+        "check the target, destination directory, and available disk space",
+    ),
     ErrorCode(
         "description_ignored",
         "`comfy workflow save --where local --description` was given a description, but the local "

comfy_cli/http.py

@@ -194,7 +194,10 @@ def build_authed_request(
     Target can't carry a stray credential into this path either.
     """
     req = urllib.request.Request(url, data=data, method=method)
-    for k, v in target_auth_headers(target).items():
+    auth_headers = target_auth_headers(target)
+    if auth_headers and urllib.parse.urlsplit(url).scheme == "http":
+        assert_safe_url(url)
+    for k, v in auth_headers.items():
         req.add_header(k, v)
     if content_type:
         req.add_header("Content-Type", content_type)

comfy_cli/schemas/nodes.json

@@ -1,7 +1,7 @@
 {
   "$schema": "https://json-schema.org/draft/2020-12/schema",
   "title": "comfy nodes *",
-  "description": "Output shape for node introspection commands (ls, show, search, upstream, downstream, path, types, categories) and the annotation-cache refresh (refresh).",
+  "description": "Output shape for node introspection commands, raw object_info snapshots, and annotation-cache refresh.",
   "type": "object",
   "properties": {
     "count": { "type": "integer" },
@@ -11,6 +11,9 @@
     "name": { "type": "string" },
     "paths": { "type": "array" },
     "types": { "type": "array" },
+    "output": { "type": "string", "description": "snapshot: destination object_info JSON file." },
+    "bytes": { "type": "integer", "description": "snapshot: bytes written." },
+    "classes": { "type": "integer", "description": "snapshot: node classes in the validated catalog." },
     "refreshed": {
       "type": "boolean",
       "description": "True only when every annotation file was re-fetched from the remote (refresh). False when any fell back to the bundled snapshot — including the intentional COMFY_CLI_NO_REMOTE_REFRESH case, which is not an error."

comfy_cli/skills/comfy/SKILL.md

@@ -355,8 +355,10 @@ partner-API node, `false` for a free open-weights one. Two nodes can share a
 display name and differ only in this flag, so read it off the row instead of
 running `nodes show` per candidate.
 
-If no local server is running and you're not signed into cloud, pass
-`--input <object_info.json>` to query against a saved dump.
+Save the live catalog for repeatable offline checks:
+`comfy nodes snapshot --output object_info.json`. If the target is later
+unreachable, pass `--input object_info.json` to node discovery or workflow
+validation.
 
 ## Models — find what's installed, with metadata
 
@@ -512,15 +514,15 @@ matches, and `total` is only how many guesses it found. Each row carries
 comfy --json workflow slots path.json   # every addressable slot, by address
 ```
 
-`workflow slots`/`set-slot`/`vary` and all `nodes` commands resolve
-object_info through the routing chain with a cached fallback — cloud-signed-in
-works with no local server. If the live fetch fails, the command still succeeds
-from cache and the envelope carries `data.stale: true` +
-`warnings[] {code: "object_info_stale"}` — treat results as possibly outdated
-re-run the command once the live fetch recovers to pick up fresh object_info.
-`comfy nodes refresh` is a different cache — it re-pulls node
-*annotations* (pack/labels/cloud_disabled) from Comfy-Org/comfy-complete, not
-object_info.
+`workflow slots`/`set-slot`/`vary` and the `nodes` query commands (`ls`,
+`show`, `search`, `upstream`, `downstream`, `path`, `types`, `categories`)
+resolve object_info through the routing chain with a cached fallback. If the
+live fetch fails, they can succeed from cache with `data.stale: true` plus
+`warnings[] {code: "object_info_stale"}`; treat results as possibly outdated
+and re-run once the target recovers. `nodes snapshot` is deliberately different:
+it requires a reachable live target so it never labels cached bytes as a fresh
+snapshot. `nodes refresh` maintains the separate annotation cache
+(pack/labels/cloud_disabled) from Comfy-Org/comfy-complete.
 
 Slot addresses are `<instance_id>.<input_name>`. Feed them to
 `workflow set-slot` / `workflow vary` in the Execution half. Works on