diff --git a/comfy_cli/command/nodes.py b/comfy_cli/command/nodes.py
index 602671869..e13585422 100644
--- a/comfy_cli/command/nodes.py
+++ b/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
 # ---------------------------------------------------------------------------
diff --git a/comfy_cli/discovery.py b/comfy_cli/discovery.py
index 82b3a693c..be367b147 100644
--- a/comfy_cli/discovery.py
+++ b/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",
diff --git a/comfy_cli/error_codes.py b/comfy_cli/error_codes.py
index 797e8a715..04c07e96d 100644
--- a/comfy_cli/error_codes.py
+++ b/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 "
diff --git a/comfy_cli/http.py b/comfy_cli/http.py
index b9598ce63..c1d06a4d4 100644
--- a/comfy_cli/http.py
+++ b/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)
diff --git a/comfy_cli/schemas/nodes.json b/comfy_cli/schemas/nodes.json
index a28268589..a44e7e799 100644
--- a/comfy_cli/schemas/nodes.json
+++ b/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."
diff --git a/comfy_cli/skills/comfy/SKILL.md b/comfy_cli/skills/comfy/SKILL.md
index 115e7a627..22ca0dff9 100644
--- a/comfy_cli/skills/comfy/SKILL.md
+++ b/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
diff --git a/tests/comfy_cli/command/test_nodes_snapshot.py b/tests/comfy_cli/command/test_nodes_snapshot.py
new file mode 100644
index 000000000..72ad5df0a
--- /dev/null
+++ b/tests/comfy_cli/command/test_nodes_snapshot.py
@@ -0,0 +1,334 @@
+from __future__ import annotations
+
+import io
+import json
+import types
+from pathlib import Path
+from typing import Any
+
+import jsonschema
+import pytest
+from typer.testing import CliRunner
+
+from comfy_cli.caller import Caller
+from comfy_cli.command import nodes as nodes_cmd
+from comfy_cli.output.renderer import OutputMode, Renderer, reset_renderer_for_testing, set_renderer
+from comfy_cli.target import Target
+
+
+@pytest.fixture(autouse=True)
+def reset_singleton():
+    reset_renderer_for_testing()
+    yield
+    reset_renderer_for_testing()
+
+
+def _force_json_renderer():
+    renderer = Renderer.resolve(
+        is_stdout_tty=False,
+        env={},
+        caller=Caller(kind="user", agentic=False, source_env=None),
+        json_flag=True,
+    )
+    renderer.mode = OutputMode.JSON
+    set_renderer(renderer)
+
+
+def _force_pretty_renderer():
+    renderer = Renderer.resolve(
+        is_stdout_tty=True,
+        env={},
+        caller=Caller(kind="user", agentic=False, source_env=None),
+        no_json_flag=True,
+    )
+    renderer.mode = OutputMode.PRETTY
+    set_renderer(renderer)
+
+
+class _ChunkedResponse:
+    def __init__(self, body: bytes, chunk_size: int = 7):
+        self._body = io.BytesIO(body)
+        self._chunk_size = chunk_size
+        self.read_sizes: list[int] = []
+        self.status = 200
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_args):
+        return False
+
+    def read(self, size: int = -1) -> bytes:
+        self.read_sizes.append(size)
+        return self._body.read(min(size, self._chunk_size) if size >= 0 else self._chunk_size)
+
+
+def _object_info() -> dict[str, Any]:
+    return {
+        "KSampler": {
+            "input": {"required": {"steps": ["INT", {"default": 20}]}},
+            "output": ["LATENT"],
+            "category": "sampling",
+        }
+    }
+
+
+def test_stream_snapshot_writes_valid_object_info_atomically(monkeypatch, tmp_path):
+    body = json.dumps(_object_info()).encode()
+    response = _ChunkedResponse(body)
+    monkeypatch.setattr(nodes_cmd, "authed_urlopen", lambda *_a, **_kw: response)
+    monkeypatch.setattr(nodes_cmd, "_OBJECT_INFO_PARSE_CHUNK_CHARS", 5)
+    target = Target(kind="local", base_url="http://127.0.0.1:8188")
+    output = tmp_path / "object_info.json"
+
+    result = nodes_cmd._stream_object_info_snapshot(target, output)
+
+    assert json.loads(output.read_text()) == _object_info()
+    assert result == {"bytes": len(body), "classes": 1}
+    assert len(response.read_sizes) > 1, "the response must be consumed incrementally"
+    assert not list(tmp_path.glob("*.tmp"))
+
+
+def test_stream_snapshot_preserves_existing_file_on_invalid_json(monkeypatch, tmp_path):
+    response = _ChunkedResponse(b'{"KSampler":')
+    monkeypatch.setattr(nodes_cmd, "authed_urlopen", lambda *_a, **_kw: response)
+    target = Target(kind="local", base_url="http://127.0.0.1:8188")
+    output = tmp_path / "object_info.json"
+    output.write_text('{"existing": true}')
+
+    with pytest.raises(ValueError, match="valid JSON"):
+        nodes_cmd._stream_object_info_snapshot(target, output)
+
+    assert output.read_text() == '{"existing": true}'
+    assert not list(tmp_path.glob("*.tmp"))
+
+
+def test_stream_snapshot_rejects_non_json_unicode_whitespace(monkeypatch, tmp_path):
+    body = '{"KSampler":\u00a0{"input":{"required":{}}}}'.encode()
+    monkeypatch.setattr(
+        nodes_cmd,
+        "authed_urlopen",
+        lambda *_a, **_kw: _ChunkedResponse(body),
+    )
+    output = tmp_path / "object_info.json"
+    output.write_text('{"existing": true}')
+
+    with pytest.raises(ValueError, match="valid JSON"):
+        nodes_cmd._stream_object_info_snapshot(Target(kind="local", base_url="http://127.0.0.1:8188"), output)
+
+    assert output.read_text() == '{"existing": true}'
+    assert not list(tmp_path.glob("*.tmp"))
+
+
+def test_stream_snapshot_accepts_exact_size_limit(monkeypatch, tmp_path):
+    body = json.dumps(_object_info()).encode()
+    monkeypatch.setattr(nodes_cmd, "_OBJECT_INFO_SNAPSHOT_MAX_BYTES", len(body))
+    monkeypatch.setattr(
+        nodes_cmd,
+        "authed_urlopen",
+        lambda *_a, **_kw: _ChunkedResponse(body),
+    )
+    output = tmp_path / "object_info.json"
+
+    result = nodes_cmd._stream_object_info_snapshot(Target(kind="local", base_url="http://127.0.0.1:8188"), output)
+
+    assert result["bytes"] == len(body)
+    assert output.is_file()
+
+
+def test_stream_snapshot_rejects_over_limit_without_replacing(monkeypatch, tmp_path):
+    body = json.dumps(_object_info()).encode()
+    monkeypatch.setattr(nodes_cmd, "_OBJECT_INFO_SNAPSHOT_MAX_BYTES", len(body) - 1)
+    monkeypatch.setattr(
+        nodes_cmd,
+        "authed_urlopen",
+        lambda *_a, **_kw: _ChunkedResponse(body),
+    )
+    output = tmp_path / "object_info.json"
+    output.write_text('{"existing": true}')
+
+    with pytest.raises(ValueError, match="snapshot limit"):
+        nodes_cmd._stream_object_info_snapshot(Target(kind="local", base_url="http://127.0.0.1:8188"), output)
+
+    assert output.read_text() == '{"existing": true}'
+    assert not list(tmp_path.glob("*.tmp"))
+
+
+def test_stream_snapshot_rejects_non_catalog_json(monkeypatch, tmp_path):
+    monkeypatch.setattr(
+        nodes_cmd,
+        "authed_urlopen",
+        lambda *_a, **_kw: _ChunkedResponse(b'{"status": "ok"}'),
+    )
+    output = tmp_path / "object_info.json"
+
+    with pytest.raises(ValueError, match="not a node catalog"):
+        nodes_cmd._stream_object_info_snapshot(Target(kind="local", base_url="http://127.0.0.1:8188"), output)
+
+    assert not output.exists()
+    assert not list(tmp_path.glob("*.tmp"))
+
+
+def test_stream_snapshot_bounds_one_catalog_entry(monkeypatch, tmp_path):
+    body = json.dumps(
+        {
+            "KSampler": {
+                "input": {"required": {}},
+                "description": "x" * 256,
+            }
+        }
+    ).encode()
+    monkeypatch.setattr(nodes_cmd, "_OBJECT_INFO_ENTRY_MAX_CHARS", 64)
+    monkeypatch.setattr(
+        nodes_cmd,
+        "authed_urlopen",
+        lambda *_a, **_kw: _ChunkedResponse(body),
+    )
+
+    with pytest.raises(ValueError, match="catalog entry exceeds"):
+        nodes_cmd._stream_object_info_snapshot(
+            Target(kind="local", base_url="http://127.0.0.1:8188"),
+            tmp_path / "object_info.json",
+        )
+
+
+@pytest.mark.parametrize(
+    ("mode", "expected_host", "expected_port"),
+    [("local", "gpu-box", 8288), ("cloud", None, None)],
+)
+def test_snapshot_target_uses_normal_routing(monkeypatch, mode, expected_host, expected_port):
+    from comfy_cli import host_port
+    from comfy_cli import where as where_module
+
+    target_kind = where_module.WhereTarget.LOCAL if mode == "local" else where_module.WhereTarget.CLOUD
+    monkeypatch.setattr(
+        where_module,
+        "resolve_default_or_exit",
+        lambda flag=None: types.SimpleNamespace(target=target_kind),
+    )
+    monkeypatch.setattr(
+        host_port,
+        "resolve_host_port",
+        lambda host, port: ("gpu-box", 8288),
+    )
+    calls: list[dict[str, Any]] = []
+
+    def fake_resolve_target(**kwargs):
+        calls.append(kwargs)
+        return Target(kind=mode, base_url="https://example.com")
+
+    monkeypatch.setattr(nodes_cmd, "resolve_target", fake_resolve_target)
+
+    nodes_cmd._resolve_snapshot_target(mode, None, None)
+
+    assert calls == [{"where": mode, "host": expected_host, "port": expected_port}]
+
+
+def test_snapshot_command_emits_written_catalog(monkeypatch, tmp_path, capsys):
+    target = Target(kind="local", base_url="http://127.0.0.1:8188")
+    monkeypatch.setattr(nodes_cmd, "_resolve_snapshot_target", lambda *_a, **_kw: target)
+    calls: list[tuple[Target, Path]] = []
+
+    def fake_snapshot(resolved_target: Target, output: Path):
+        calls.append((resolved_target, output))
+        output.write_text(json.dumps(_object_info()))
+        return {"bytes": output.stat().st_size, "classes": 1}
+
+    monkeypatch.setattr(nodes_cmd, "_stream_object_info_snapshot", fake_snapshot)
+    output = tmp_path / "object_info.json"
+    _force_json_renderer()
+
+    result = CliRunner().invoke(nodes_cmd.app, ["snapshot", "--output", str(output)], standalone_mode=False)
+    captured = capsys.readouterr().out or result.stdout
+    envelope = json.loads(captured.strip().splitlines()[-1])
+
+    assert result.exit_code == 0
+    assert envelope["ok"] is True
+    assert envelope["data"]["output"] == str(output)
+    assert envelope["data"]["classes"] == 1
+    assert calls == [(target, output)]
+    schema = json.loads((Path(nodes_cmd.__file__).parent.parent / "schemas" / "nodes.json").read_text())
+    jsonschema.validate(instance=envelope["data"], schema=schema)
+
+
+def test_snapshot_command_sanitizes_pretty_output(monkeypatch, tmp_path, capsys):
+    target = Target(kind="local", base_url="http://127.0.0.1:8188")
+    monkeypatch.setattr(nodes_cmd, "_resolve_snapshot_target", lambda *_a, **_kw: target)
+
+    def fake_snapshot(_target: Target, output: Path):
+        output.write_text(json.dumps(_object_info()))
+        return {"bytes": output.stat().st_size, "classes": 1}
+
+    monkeypatch.setattr(nodes_cmd, "_stream_object_info_snapshot", fake_snapshot)
+    output = tmp_path / "[red]snapshot.json"
+    _force_pretty_renderer()
+
+    result = CliRunner().invoke(
+        nodes_cmd.app,
+        ["snapshot", "--output", str(output)],
+        standalone_mode=False,
+    )
+    captured = capsys.readouterr().out or result.stdout
+
+    assert result.exit_code == 0
+    assert "[red]snapshot.json" in captured
+
+
+def test_snapshot_command_reports_invalid_where(tmp_path, capsys):
+    _force_json_renderer()
+
+    result = CliRunner().invoke(
+        nodes_cmd.app,
+        [
+            "snapshot",
+            "--output",
+            str(tmp_path / "object_info.json"),
+            "--where",
+            "somewhere",
+        ],
+    )
+    captured = capsys.readouterr().out or result.stdout
+    envelope = json.loads(captured.strip().splitlines()[-1])
+
+    assert result.exit_code == 1
+    assert envelope["ok"] is False
+    assert envelope["error"]["code"] == "where_invalid"
+
+
+def test_snapshot_command_maps_unresolvable_home_to_envelope(monkeypatch, capsys):
+    target = Target(kind="local", base_url="http://127.0.0.1:8188")
+    monkeypatch.setattr(nodes_cmd, "_resolve_snapshot_target", lambda *_a, **_kw: target)
+    _force_json_renderer()
+
+    result = CliRunner().invoke(
+        nodes_cmd.app,
+        ["snapshot", "--output", "~comfy-cli-user-that-does-not-exist/catalog.json"],
+    )
+    captured = capsys.readouterr().out or result.stdout
+    envelope = json.loads(captured.strip().splitlines()[-1])
+
+    assert result.exit_code == 1
+    assert envelope["ok"] is False
+    assert envelope["error"]["code"] == "nodes_snapshot_failed"
+
+
+def test_snapshot_command_maps_fetch_failure_to_envelope(monkeypatch, tmp_path, capsys):
+    target = Target(kind="local", base_url="http://127.0.0.1:8188")
+    monkeypatch.setattr(nodes_cmd, "_resolve_snapshot_target", lambda *_a, **_kw: target)
+
+    def fail_snapshot(*_args, **_kwargs):
+        raise OSError("connection lost")
+
+    monkeypatch.setattr(nodes_cmd, "_stream_object_info_snapshot", fail_snapshot)
+    _force_json_renderer()
+
+    result = CliRunner().invoke(
+        nodes_cmd.app,
+        ["snapshot", "--output", str(tmp_path / "object_info.json")],
+    )
+    captured = capsys.readouterr().out or result.stdout
+    envelope = json.loads(captured.strip().splitlines()[-1])
+
+    assert result.exit_code == 1
+    assert envelope["ok"] is False
+    assert envelope["error"]["code"] == "nodes_snapshot_failed"
diff --git a/tests/comfy_cli/test_http.py b/tests/comfy_cli/test_http.py
index e00898f14..e1546b92a 100644
--- a/tests/comfy_cli/test_http.py
+++ b/tests/comfy_cli/test_http.py
@@ -102,6 +102,14 @@ def test_bearer_when_only_auth_token():
     assert req.get_header("X-api-key") is None
 
 
+def test_credential_rejected_on_plaintext_non_loopback_url():
+    with pytest.raises(ValueError, match="non-https, non-loopback"):
+        build_authed_request(
+            "http://api.example.com/object_info",
+            _target(auth_token="secret"),
+        )
+
+
 def test_no_auth_header_when_uncredentialed():
     """A local (uncredentialed) target gets no credential header."""
     req = build_authed_request("https://x/thing", _target())