From f4ffc519d5a53393b6f577138fa1f449e5ad8406 Mon Sep 17 00:00:00 2001 From: Matt Miller Date: Thu, 23 Jul 2026 16:10:58 -0700 Subject: [PATCH 1/2] fix(jobs): scope `jobs ls` state-file rows to the resolved --where target MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `--where` only routed the SERVER query: `_gather_local_state_files` read every state file in the jobs state dir regardless of each state's `where`, so `comfy --json --where local jobs ls` kept surfacing cloud jobs from previous runs (state files never expire). Agents wrapping the CLI with a pinned `--where local` were getting cloud rows indefinitely. - `_gather_local_state_files` takes a `where` filter; a missing/empty `state.where` reads as "local". `None` keeps the unfiltered union view. - `jobs ls` resolves the target once and scopes state rows to it; the new `--all` flag restores the union. `--orphaned` stays unfiltered — watcher cleanup is where-agnostic. - `--watch` threads the same scope so the live table matches `jobs ls`. - Payload gains `scope` ("local" / "cloud" / "all") so callers can see which view they got. Additive; envelope/1 is unchanged. - `server_rows` are deliberately not filtered: they are already scoped by whichever backend was queried. --- comfy_cli/command/jobs.py | 78 ++++++++++++--- comfy_cli/schemas/jobs.json | 5 + comfy_cli/skills/comfy/SKILL.md | 7 ++ tests/comfy_cli/jobs/test_jobs.py | 154 ++++++++++++++++++++++++++++++ 4 files changed, 232 insertions(+), 12 deletions(-) diff --git a/comfy_cli/command/jobs.py b/comfy_cli/command/jobs.py index 9d6db398..1fe14094 100644 --- a/comfy_cli/command/jobs.py +++ b/comfy_cli/command/jobs.py @@ -164,7 +164,7 @@ class JobRow: updated_at: str | None = None # ISO timestamp, set for state-file rows -def _gather_local_state_files(*, limit: int, orphaned_only: bool = False) -> list[JobRow]: +def _gather_local_state_files(*, limit: int, orphaned_only: bool = False, where: str | None = None) -> list[JobRow]: """Read every state file in the jobs state dir → JobRow. This is the canonical "what did *I* submit via this CLI" view — @@ -174,6 +174,12 @@ def _gather_local_state_files(*, limit: int, orphaned_only: bool = False) -> lis When ``orphaned_only`` is True, return only rows whose state file has ``error.code == "watcher_crashed"`` — jobs where the background watcher died and was reaped. Useful for cleanup. + + ``where`` scopes the rows to one routing target (``"local"`` / + ``"cloud"``) so a ``--where local`` listing can't surface cloud jobs + submitted in an earlier run. A state file whose ``where`` is missing or + empty counts as ``"local"``. ``None`` (the default) keeps the unfiltered + union view — used by ``jobs ls --all`` and ``--orphaned``. """ import re as _re @@ -208,6 +214,11 @@ def _gather_local_state_files(*, limit: int, orphaned_only: bool = False) -> lis } state.watcher_pid = None jobs_state.write(state) + # Scope to the resolved --where target. Done *after* the stale-watcher + # reap above so cleanup stays where-agnostic no matter which view the + # caller asked for. + if where is not None and (state.where or "local") != where: + continue if orphaned_only: err = state.error or {} if not (isinstance(err, dict) and err.get("code") == "watcher_crashed"): @@ -220,7 +231,10 @@ def _gather_local_state_files(*, limit: int, orphaned_only: bool = False) -> lis elapsed_seconds=None, workflow_size=None, outputs=len(state.outputs or []), - where=state.where, + # Same missing/empty -> "local" reading the filter above uses, + # so a row can never be scoped as local yet report `where: null` + # (JobRow.where is typed `str` and defaults to "local"). + where=state.where or "local", workflow_path=state.workflow, updated_at=state.updated_at, ) @@ -348,7 +362,13 @@ def _safe_queue_entry(entry: Any) -> tuple[str, Any]: return ("?", None) -@app.command("ls", help="List jobs: locally-tracked async submits + server queue/history.") +@app.command( + "ls", + help=( + "List jobs: locally-tracked async submits + server queue/history, " + "scoped to the resolved --where target (use --all for every target)." + ), +) @tracking.track_command("jobs") def ls_cmd( host: Annotated[str | None, typer.Option(help="Server host (defaults to background or 127.0.0.1).")] = None, @@ -378,6 +398,14 @@ def ls_cmd( ), ), ] = False, + all_wheres: Annotated[ + bool, + typer.Option( + "--all", + show_default=False, + help="Show state-file rows for every target, not just the resolved --where.", + ), + ] = False, watch: Annotated[ bool, typer.Option( @@ -389,6 +417,11 @@ def ls_cmd( ): renderer = get_renderer() + # Resolve the routing target once: per-command --where flag > COMFY_WHERE + # env (how the top-level `comfy --where` arrives) > config default. Both + # the server query and the state-file scope key off this single decision. + target_where = "cloud" if _is_cloud(where) else "local" + if watch: if not renderer.is_pretty(): renderer.error( @@ -397,20 +430,33 @@ def ls_cmd( hint="drop --json, or run `while true; do comfy --json jobs ls; sleep 2; done`", ) raise typer.Exit(code=1) - _watch_ls(host=host, port=port, limit=limit, where=where, local_only=local_only) + _watch_ls( + host=host, + port=port, + limit=limit, + where=where, + local_only=local_only, + state_where=None if all_wheres else target_where, + ) return - state_rows = _gather_local_state_files(limit=limit, orphaned_only=orphaned) - # --orphaned only makes sense for state files (the server doesn't know # whether a watcher crashed), so skip the server query in that mode. if orphaned: local_only = True + # State-file rows are scoped to the resolved target, so a `--where local` + # listing can't surface cloud jobs from an earlier run; `--all` restores + # the union view. `--orphaned` stays unfiltered — watcher cleanup is + # where-agnostic. `server_rows` are never filtered: they are already + # scoped by which backend we queried. + state_where = None if (all_wheres or orphaned) else target_where + state_rows = _gather_local_state_files(limit=limit, orphaned_only=orphaned, where=state_where) + server_rows: list[JobRow] = [] h, p = _resolve_host_port(host, port) if not local_only: - if _is_cloud(where): + if target_where == "cloud": try: cloud_preflight_or_exit() client = _cloud_client() @@ -430,12 +476,15 @@ def ls_cmd( rows = _merge_jobs(state_rows, server_rows)[:limit] if renderer.is_pretty(): - _render_jobs_pretty(rows, host=h if not _is_cloud(where) else "cloud.comfy.org", port=p) + _render_jobs_pretty(rows, host=h if target_where != "cloud" else "cloud.comfy.org", port=p) renderer.emit( { "host": h, "port": p, - "where": "cloud" if _is_cloud(where) else "local", + "where": target_where, + # Which state-file view the caller got: the resolved target, or + # "all" when the union view was requested (--all/--orphaned). + "scope": state_where or "all", "count": len(rows), "jobs": [_row_to_dict(r) for r in rows], }, @@ -443,8 +492,13 @@ def ls_cmd( ) -def _watch_ls(*, host, port, limit, where, local_only): - """Rich Live refresh of the jobs table every 2s until Ctrl-C.""" +def _watch_ls(*, host, port, limit, where, local_only, state_where=None): + """Rich Live refresh of the jobs table every 2s until Ctrl-C. + + ``state_where`` scopes the state-file rows exactly as the one-shot path + does (``None`` = the unfiltered union view, i.e. ``--all``), so the live + table shows the same jobs as ``jobs ls``. + """ import time from rich.live import Live @@ -457,7 +511,7 @@ def _watch_ls(*, host, port, limit, where, local_only): h, p = _resolve_host_port(host, port) def build_table() -> Table: - state_rows = _gather_local_state_files(limit=limit) + state_rows = _gather_local_state_files(limit=limit, where=state_where) server_rows: list[JobRow] = [] if not local_only: try: diff --git a/comfy_cli/schemas/jobs.json b/comfy_cli/schemas/jobs.json index ef366a64..1b76d67a 100644 --- a/comfy_cli/schemas/jobs.json +++ b/comfy_cli/schemas/jobs.json @@ -27,6 +27,11 @@ "description": "Cloud status/watch terminal: artifact URLs grouped by blueprint foreach item (via the compose item_map on the job state file). Empty object when no item_map exists.", "additionalProperties": {"type": "array", "items": {"type": "string"}} }, + "scope": { + "type": "string", + "enum": ["local", "cloud", "all"], + "description": "`jobs ls`: which state-file rows the listing includes — only those matching the resolved --where target, or \"all\" targets when `--all` (or `--orphaned`) was passed." + }, "elapsed_seconds": {"type": ["number", "null"]}, "completed_nodes": {"type": "array", "items": {"type": "string"}}, "count": {"type": "integer"}, diff --git a/comfy_cli/skills/comfy/SKILL.md b/comfy_cli/skills/comfy/SKILL.md index 51d2303e..d0fc65c6 100644 --- a/comfy_cli/skills/comfy/SKILL.md +++ b/comfy_cli/skills/comfy/SKILL.md @@ -714,10 +714,17 @@ bad wiring — before burning cloud compute. ```bash comfy --json jobs ls # merged: local state files + server queue +comfy --json jobs ls --all # ...every target, not just the resolved --where comfy --json jobs status comfy --json jobs watch # blocks until terminal; emits NDJSON with --json-stream ``` +`jobs ls` is scoped to the resolved `--where` target: a `--where local` +listing shows local state-file rows only, a `--where cloud` listing cloud +ones. The payload's `scope` field says which view you got (`local` / +`cloud` / `all`); pass `--all` for the union of everything this CLI +submitted. + Terminal envelopes (`run --wait`, `jobs status`, `jobs watch`) carry the flat `outputs` list plus grouped views of the same artifacts: `outputs_by_node: {node_id: [url]}` always, and `outputs_by_item: diff --git a/tests/comfy_cli/jobs/test_jobs.py b/tests/comfy_cli/jobs/test_jobs.py index 5ed15898..4c3cef09 100644 --- a/tests/comfy_cli/jobs/test_jobs.py +++ b/tests/comfy_cli/jobs/test_jobs.py @@ -13,6 +13,7 @@ from pathlib import Path import pytest +import typer from comfy_cli.command import jobs as jobs_mod @@ -319,6 +320,159 @@ def test_orphaned_flag_visible_in_help(): assert "--orphaned" in _command_flags("jobs", "ls") +def test_all_flag_visible_in_help(): + """``--all`` is the escape hatch back to the union view — agents must be + able to discover it from `comfy --help-json`.""" + assert "--all" in _command_flags("jobs", "ls") + + +# --------------------------------------------------------------------------- +# `jobs ls` state-file rows are scoped to the resolved --where target +# --------------------------------------------------------------------------- + + +def _seed_three_targets() -> Path: + """Seed the (already isolated) state dir with one local, one cloud and + one legacy (``where`` present but null) job. Returns the dir.""" + from comfy_cli import jobs_state + + state_dir = jobs_state.state_dir() + _write_state(state_dir, "job-local", where="local", status="completed") + _write_state(state_dir, "job-cloud", where="cloud", status="completed") + # Legacy files predate the cloud target. ``jobs_state.read`` requires the + # key to be present (it's a non-defaulted dataclass field), so the + # reachable legacy shape is a null/empty value, not an absent one — those + # must read as "local". + _write_state(state_dir, "job-legacy", where=None, status="completed") + return state_dir + + +def test_gather_local_state_files_filters_by_where(): + """The ``where`` kwarg scopes state-file rows; ``None`` keeps the union.""" + _seed_three_targets() + + local_ids = {r.prompt_id for r in jobs_mod._gather_local_state_files(limit=100, where="local")} + assert local_ids == {"job-local", "job-legacy"}, f"missing/empty where must count as local; got {local_ids}" + + cloud_ids = {r.prompt_id for r in jobs_mod._gather_local_state_files(limit=100, where="cloud")} + assert cloud_ids == {"job-cloud"} + + union_ids = {r.prompt_id for r in jobs_mod._gather_local_state_files(limit=100)} + assert union_ids == {"job-local", "job-cloud", "job-legacy"} + + +def test_gather_local_state_files_drops_file_with_no_where_key(): + """Belt-and-braces on the legacy shape: a state file that omits ``where`` + entirely never reaches the filter — ``jobs_state.read`` already rejects it + as truncated — so it can't leak into a scoped *or* an --all listing.""" + from comfy_cli import jobs_state + + state_dir = jobs_state.state_dir() + (state_dir / "job-nowhere.json").write_text( + json.dumps( + { + "prompt_id": "job-nowhere", + "client_id": "c", + "workflow": "/tmp/x.json", + "status": "completed", + } + ) + ) + assert jobs_state.read("job-nowhere") is None + assert jobs_mod._gather_local_state_files(limit=100) == [] + + +def _ls_payload(capsys, **kwargs) -> dict: + """Run ``jobs ls`` in-process under a JSON renderer, return its data dict.""" + from comfy_cli.caller import Caller + from comfy_cli.output.renderer import OutputMode, Renderer, set_renderer + + r = Renderer.resolve( + is_stdout_tty=False, + env={}, + caller=Caller(kind="user", agentic=False, source_env=None), + json_flag=True, + ) + r.mode = OutputMode.JSON + set_renderer(r) + # 65431 is a port nothing listens on — the server query degrades to the + # state-file view, which is exactly the leak this scoping guards. + kwargs.setdefault("host", "127.0.0.1") + kwargs.setdefault("port", 65431) + jobs_mod.ls_cmd(**kwargs) + env = _last_json(capsys.readouterr().out) + assert env["ok"] is True + return env["data"] + + +def _ls_ids(payload: dict) -> set[str]: + return {j["prompt_id"] for j in payload["jobs"]} + + +def test_ls_default_scopes_state_rows_to_local(capsys, monkeypatch): + """Acceptance: with a cloud job on disk, a local `jobs ls` has no cloud rows.""" + _seed_three_targets() + monkeypatch.delenv("COMFY_WHERE", raising=False) + + data = _ls_payload(capsys, limit=100) + assert data["where"] == "local" + assert data["scope"] == "local" + assert _ls_ids(data) == {"job-local", "job-legacy"} + # Acceptance: no cloud rows at all — and the legacy row reports the target + # it was scoped under rather than a bare null. + assert {j["where"] for j in data["jobs"]} == {"local"} + + +def test_ls_where_cloud_scopes_state_rows_to_cloud(capsys, monkeypatch): + """Acceptance: `jobs ls --where cloud` shows no local state rows.""" + _seed_three_targets() + monkeypatch.setattr(jobs_mod, "_is_cloud", lambda w: True) + + def _preflight_fails(): + raise typer.Exit(code=1) + + # Cloud unreachable/unauthed: the command falls through to the state-file + # view, which must still be cloud-scoped. + monkeypatch.setattr(jobs_mod, "cloud_preflight_or_exit", _preflight_fails) + + data = _ls_payload(capsys, limit=100, where="cloud") + assert data["where"] == "cloud" + assert data["scope"] == "cloud" + assert _ls_ids(data) == {"job-cloud"} + + +def test_ls_all_restores_the_union_view(capsys, monkeypatch): + """Acceptance: ``--all`` brings every target's state rows back.""" + _seed_three_targets() + monkeypatch.delenv("COMFY_WHERE", raising=False) + + data = _ls_payload(capsys, limit=100, all_wheres=True) + assert data["where"] == "local", "--all widens the state-file scope, not the server query" + assert data["scope"] == "all" + assert _ls_ids(data) == {"job-local", "job-cloud", "job-legacy"} + + +def test_ls_orphaned_stays_unfiltered(capsys, monkeypatch): + """``--orphaned`` keeps the union view — watcher cleanup is where-agnostic, + so a crashed cloud watcher is still reapable from a default (local) ls.""" + from comfy_cli import jobs_state + + state_dir = jobs_state.state_dir() + crashed = { + "code": "watcher_crashed", + "message": "Background watcher (pid 99999) is no longer running.", + "hint": "re-submit the workflow", + } + _write_state(state_dir, "orphan-local", where="local", status="error", error=crashed) + _write_state(state_dir, "orphan-cloud", where="cloud", status="error", error=crashed) + _write_state(state_dir, "healthy-cloud", where="cloud", status="completed") + monkeypatch.delenv("COMFY_WHERE", raising=False) + + data = _ls_payload(capsys, limit=100, orphaned=True) + assert data["scope"] == "all" + assert _ls_ids(data) == {"orphan-local", "orphan-cloud"} + + # --------------------------------------------------------------------------- # --where routing — top-level flag must be honored, not just per-command # --------------------------------------------------------------------------- From c7eabbfa83cde8a3fe19199c1ddf7d00eecbc308 Mon Sep 17 00:00:00 2001 From: Matt Miller Date: Thu, 23 Jul 2026 16:29:20 -0700 Subject: [PATCH 2/2] fix(jobs): make `jobs ls --watch` honor --orphaned (Cursor review) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `--watch` branch returned before `if orphaned: local_only = True`, and `_watch_ls` never received `orphaned`, so `jobs ls --watch --orphaned` still queried the server, listed every state file instead of only crashed-watcher rows, and scoped state files to the resolved target instead of the union — diverging from the one-shot behavior this PR documents. Hoist the `orphaned` handling and the `state_where` decision above the `--watch` branch and thread `orphaned_only` through `_watch_ls` into `_gather_local_state_files`, so the live table applies exactly the same filters as `jobs ls`. Tests cover the kwargs the watch path is driven with (default / --all / --where cloud / --orphaned) plus `_watch_ls` actually forwarding them. Co-Authored-By: Claude Opus 4.8 --- comfy_cli/command/jobs.py | 38 ++++++++++-------- tests/comfy_cli/jobs/test_jobs.py | 65 +++++++++++++++++++++++++++++++ 2 files changed, 87 insertions(+), 16 deletions(-) diff --git a/comfy_cli/command/jobs.py b/comfy_cli/command/jobs.py index 1fe14094..80374f97 100644 --- a/comfy_cli/command/jobs.py +++ b/comfy_cli/command/jobs.py @@ -422,6 +422,21 @@ def ls_cmd( # the server query and the state-file scope key off this single decision. target_where = "cloud" if _is_cloud(where) else "local" + # --orphaned only makes sense for state files (the server doesn't know + # whether a watcher crashed), so skip the server query in that mode. + if orphaned: + local_only = True + + # State-file rows are scoped to the resolved target, so a `--where local` + # listing can't surface cloud jobs from an earlier run; `--all` restores + # the union view. `--orphaned` stays unfiltered — watcher cleanup is + # where-agnostic. `server_rows` are never filtered: they are already + # scoped by which backend we queried. + # + # Both decisions are made *before* the --watch branch so the live table + # applies exactly the same filters as the one-shot listing. + state_where = None if (all_wheres or orphaned) else target_where + if watch: if not renderer.is_pretty(): renderer.error( @@ -436,21 +451,11 @@ def ls_cmd( limit=limit, where=where, local_only=local_only, - state_where=None if all_wheres else target_where, + state_where=state_where, + orphaned_only=orphaned, ) return - # --orphaned only makes sense for state files (the server doesn't know - # whether a watcher crashed), so skip the server query in that mode. - if orphaned: - local_only = True - - # State-file rows are scoped to the resolved target, so a `--where local` - # listing can't surface cloud jobs from an earlier run; `--all` restores - # the union view. `--orphaned` stays unfiltered — watcher cleanup is - # where-agnostic. `server_rows` are never filtered: they are already - # scoped by which backend we queried. - state_where = None if (all_wheres or orphaned) else target_where state_rows = _gather_local_state_files(limit=limit, orphaned_only=orphaned, where=state_where) server_rows: list[JobRow] = [] @@ -492,12 +497,13 @@ def ls_cmd( ) -def _watch_ls(*, host, port, limit, where, local_only, state_where=None): +def _watch_ls(*, host, port, limit, where, local_only, state_where=None, orphaned_only=False): """Rich Live refresh of the jobs table every 2s until Ctrl-C. ``state_where`` scopes the state-file rows exactly as the one-shot path - does (``None`` = the unfiltered union view, i.e. ``--all``), so the live - table shows the same jobs as ``jobs ls``. + does (``None`` = the unfiltered union view, i.e. ``--all``), and + ``orphaned_only`` mirrors ``--orphaned``, so the live table shows the same + jobs as ``jobs ls``. """ import time @@ -511,7 +517,7 @@ def _watch_ls(*, host, port, limit, where, local_only, state_where=None): h, p = _resolve_host_port(host, port) def build_table() -> Table: - state_rows = _gather_local_state_files(limit=limit, where=state_where) + state_rows = _gather_local_state_files(limit=limit, where=state_where, orphaned_only=orphaned_only) server_rows: list[JobRow] = [] if not local_only: try: diff --git a/tests/comfy_cli/jobs/test_jobs.py b/tests/comfy_cli/jobs/test_jobs.py index 4c3cef09..93c9d178 100644 --- a/tests/comfy_cli/jobs/test_jobs.py +++ b/tests/comfy_cli/jobs/test_jobs.py @@ -473,6 +473,71 @@ def test_ls_orphaned_stays_unfiltered(capsys, monkeypatch): assert _ls_ids(data) == {"orphan-local", "orphan-cloud"} +def _watch_kwargs(monkeypatch, **kwargs) -> dict: + """Run ``jobs ls --watch`` under a pretty renderer with ``_watch_ls`` + stubbed, and return the kwargs the live path would have been driven with.""" + from comfy_cli.output.renderer import OutputMode, Renderer, set_renderer + + set_renderer(Renderer(mode=OutputMode.PRETTY, command="jobs ls")) + seen: dict = {} + monkeypatch.setattr(jobs_mod, "_watch_ls", lambda **kw: seen.update(kw)) + jobs_mod.ls_cmd(host="127.0.0.1", port=65431, watch=True, **kwargs) + return seen + + +def test_ls_watch_mirrors_one_shot_scope(monkeypatch): + """``--watch`` must apply the *same* state-file filters as the one-shot + listing: the resolved target by default, the union under ``--all``.""" + monkeypatch.delenv("COMFY_WHERE", raising=False) + + assert _watch_kwargs(monkeypatch)["state_where"] == "local" + assert _watch_kwargs(monkeypatch, all_wheres=True)["state_where"] is None + + monkeypatch.setattr(jobs_mod, "_is_cloud", lambda w: True) + assert _watch_kwargs(monkeypatch, where="cloud")["state_where"] == "cloud" + + +def test_ls_watch_threads_orphaned(monkeypatch): + """``jobs ls --watch --orphaned`` must restrict to crashed-watcher rows, + keep the union scope, and skip the server query — exactly like one-shot. + The `if orphaned` handling used to sit *below* the --watch early return.""" + monkeypatch.delenv("COMFY_WHERE", raising=False) + + seen = _watch_kwargs(monkeypatch, orphaned=True) + assert seen["orphaned_only"] is True + assert seen["state_where"] is None, "--orphaned is where-agnostic in watch too" + assert seen["local_only"] is True, "the server can't know a watcher crashed" + + +def test_watch_build_table_applies_orphaned_and_scope(monkeypatch): + """The stubbed kwargs above are only useful if ``_watch_ls`` actually + forwards them to the gatherer.""" + from comfy_cli.output.renderer import OutputMode, Renderer, set_renderer + + class _Stop(Exception): + pass + + set_renderer(Renderer(mode=OutputMode.PRETTY, command="jobs ls")) + seen: dict = {} + monkeypatch.setattr(jobs_mod, "_gather_local_state_files", lambda **kw: seen.update(kw) or []) + # Bail out of the first table build rather than looping forever. `_Stop` is + # not KeyboardInterrupt on purpose — `_watch_ls` swallows that one. + monkeypatch.setattr(jobs_mod, "_merge_jobs", lambda *_a, **_k: (_ for _ in ()).throw(_Stop())) + + with pytest.raises(_Stop): + jobs_mod._watch_ls( + host="127.0.0.1", + port=65431, + limit=100, + where="local", + local_only=True, + state_where=None, + orphaned_only=True, + ) + assert seen["orphaned_only"] is True + assert seen["where"] is None + + # --------------------------------------------------------------------------- # --where routing — top-level flag must be honored, not just per-command # ---------------------------------------------------------------------------