diff --git a/comfy_cli/command/jobs.py b/comfy_cli/command/jobs.py index 96617954..1d6457ba 100644 --- a/comfy_cli/command/jobs.py +++ b/comfy_cli/command/jobs.py @@ -170,7 +170,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 — @@ -180,6 +180,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 @@ -214,6 +220,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"): @@ -226,7 +237,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, ) @@ -354,7 +368,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, @@ -384,6 +404,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( @@ -395,6 +423,26 @@ 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" + + # --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( @@ -403,20 +451,23 @@ 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=state_where, + orphaned_only=orphaned, + ) 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_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() @@ -436,12 +487,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], }, @@ -449,8 +503,14 @@ 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, 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``), and + ``orphaned_only`` mirrors ``--orphaned``, so the live table shows the same + jobs as ``jobs ls``. + """ import time from rich.live import Live @@ -463,7 +523,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, orphaned_only=orphaned_only) 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 7b2eee42..cde810e6 100644 --- a/comfy_cli/skills/comfy/SKILL.md +++ b/comfy_cli/skills/comfy/SKILL.md @@ -745,10 +745,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 f3efe364..3329c0cd 100644 --- a/tests/comfy_cli/jobs/test_jobs.py +++ b/tests/comfy_cli/jobs/test_jobs.py @@ -15,6 +15,7 @@ import pytest import requests +import typer from comfy_cli.command import jobs as jobs_mod @@ -433,6 +434,224 @@ 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"} + + +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 # ---------------------------------------------------------------------------