fix(run): emit the queued NDJSON event only after the job state file is persisted

[mattmillerai]

STACKED — merging lands on matt/be-6039-run-json-cloud-event-contract (owned by #656, @mattmillerai), NOT main. #656 introduces the contractual cloud queued event this PR relocates and touches the same lines. Review the net diff (3 files); do not treat this as ready to merge to main until #656 lands, at which point GitHub retargets this PR.

ELI-5

comfy run --json-stream prints a queued line meaning "the server took your job." It printed that line before writing the little file on disk that comfy jobs ls / comfy jobs status read. So a script that saw queued and immediately asked "where's my job?" could be told it doesn't exist.

Worse on cloud: if the thing reading the stream had already walked away (… | head -n1), printing that line blew up — and it blew up before the file was written. The CLI's global broken-pipe handling then exited 0, so you got a cloud job you're paying for with no record of it anywhere: no journal entry, no state file, no background watcher.

This moves the line to after the record is written, on all four paths (local/cloud × --wait/--no-wait). Same line, same fields — just later.

What changed

Cloud (execute_cloud) — the queued emit moves out of the shared post-submit position into each branch:

• --no-wait: after jobs_state.write + _journal_run + _spawn_watcher, before the pretty print / envelope.
• --wait: after jobs_state.write + _journal_run, and outside the polling try — so none of its handlers can see a BrokenPipeError from this emit and rewrite state for it. It propagates to click/__main__ (silent exit 0) with the durable record already on disk, which is the correct semantics.

Both branches call one local closure so the six-field payload can't drift between them.

Local (WorkflowExecution.queue() + execute) — queue() no longer emits; it stashes self.validation_warnings (new, defaults []) and the caller emits the identical field set via _emit_queued():

• --wait: after _write_state(wait_state), before watch_execution() — the documented prompt_preview → queued → node events order still holds.
• --no-wait: after jobs_state.write + _spawn_watcher, before the pretty print / envelope.

The load-bearing guard. Both local emit points sit inside the try whose handler is except (WebSocketException, ConnectionError, OSError). BrokenPipeError is a ConnectionError subclass, so without a guard a closed stdout at the relocated emit would be misreported as ws_disconnected and rewrite the just-persisted healthy running record to error/server_died. The handler now re-raises BrokenPipeError as its first statement — the same principle already documented where completed_payload is emitted outside the try.

Docs — docs/json-output.md#queued now states queued is emitted after the submit succeeds and after the state file is persisted (and after the watcher spawn attempt on the async paths), so a consumer may rely on comfy jobs status / jobs ls the moment it reads the line. The field set is unchanged.

Tests

tests/comfy_cli/command/test_run_json.py, ~9 new cases:

• Ordering spies, all four target × wait-mode combinations: jobs_state.write and the queued line each append to a shared log via a stream tap; state_write must precede queued_emit.
• Broken pipe at the emit, async paths (local + cloud): the stream raises BrokenPipeError on the queued line only; asserts jobs_state.write and _spawn_watcher both ran first, and that the propagated exception is BrokenPipeError itself (not swallowed, not converted) so __main__ can produce its exit 0.
• Broken pipe at the emit, local --wait: asserts the persisted record still reads status == "running" with no error — pinning the handler guard.
• queue() stashes without emitting, including the non-empty validation_warnings case (200 + node_errors); the existing test_validation_warnings_on_200_with_partial_node_errors still asserts the same warnings surface on the queued event in the caller's output, unchanged.

Mutation-checked both ways: dropping the handler guard fails the two broken-pipe tests (with exactly the ws_disconnected + state-rewrite symptom described above); hoisting either emit back above its state write fails the ordering tests.

Verification: full pytest green, ruff check / ruff format --check clean on every touched file. (The repo has 17 pre-existing UP038 findings and one unformatted file, tests/comfy_cli/command/github/test_pr.py — none in files this PR touches.)

Judgment calls / residual risk for the reviewer

1. Stacked rather than blocked. The ticket said to hand this back if fix(run): conform the cloud NDJSON stream to the documented event contract #656 was still open. fix(run): conform the cloud NDJSON stream to the documented event contract #656 is open but has a buildable branch, so this stacks on it per the always-stack convention instead of idling. The net diff here is only this change.

2. The guard widens slightly beyond the relocated emit — it catches any BrokenPipeError in that try, not just one from the queued line. Two consequences worth a reviewer's eye, neither of which any test in this PR pins:

   • A BrokenPipeError from the WebSocket handshake in connect() (before any prompt exists) now propagates as a silent exit 0 instead of ws_disconnected / exit 1. Narrow — a handshake failure normally surfaces as ConnectionRefusedError, a timeout, or a WebSocketException, and EPIPE on our own stdout is by far the likelier source — but it is a behavior delta.
   • A BrokenPipeError mid-watch_execution now exits 0 leaving a running record rather than marking server_died. Mid-watch broken-pipe behavior was explicitly scoped out of this change, but the guard necessarily covers it; a WS read failure yields ConnectionResetError, not EPIPE, so in practice this path is also a closed stdout and exiting 0 is the intended semantics. Happy to narrow the guard to the emit specifically if you'd rather keep those two cases exactly as they were.

3. Negative-claim falsification: not applicable. This diff adds no capability denial — no "not supported"/"unavailable" string, no new throw/deny dead-end, no test flipped to assert one. The single added raise is a re-raise that restores the CLI's already-documented broken-pipe behavior (silent exit 0) on a path that was swallowing it.

[coderabbitai]

Important

Review skipped

Auto reviews are disabled on base/target branches other than the default branch.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: d94372e6-bf7e-4bf3-be19-d94fae2f4637

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

Use the checkbox below for a quick retry:

• 🔍 Trigger review

---

_{Comment @coderabbitai help to get the list of available commands.}

[github-actions]

🔍 Cursor Review — Consolidated panel

Triggered by @mattmillerai.

Found 9 finding(s).

Severity	Count
🟠 High	4
🟡 Medium	3
🟢 Low	1
⚪ Nit	1

Panel: 8/8 reviewers contributed findings.

[github-actions]

🟠 High — isinstance(e, BrokenPipeError) cannot tell a closed stdout from a genuine EPIPE on another fd: this try wraps execution.connect() and execution.watch_execution(), and websocket-client writes pong/close frames back to the server during recv(), so a server that vanishes mid-run can surface here as BrokenPipeError and get laundered into click's silent exit 0 with no ws_disconnected envelope and the record left non-terminal. It also short-circuits the token.is_set() check below, so a Ctrl-C that ends in EPIPE skips _mark_cancelled and the cancelled/130 envelope. Gate on stdout ownership (e.g. check the renderer's stream is the failing one) rather than the exception type alone. Raised by 4 of 8 reviewers (claude-opus-5-thinking-max adversarial, claude-opus-5-thinking-max edge-case, gemini-3.1-pro adversarial, gpt-5.6-sol-max edge-case).

[mattmillerai]

Fixed in the follow-up commit — you're right that the exception type can't attribute the failing fd, and the short-circuit was the worse half of it.

The bail-out now gates on stdout ownership rather than the type: Renderer._write_json_line flags _stream_broken = True when one of its own machine_stream writes raises OSError, and the handler checks renderer.stream_broken. A BrokenPipeError that websocket-client raised from its own pong/close sends never sets that flag, so it now falls through to the ws_disconnected path with the server_died record write intact.

I also moved the token.is_set() check above the bail-out, so a Ctrl-C that ends in EPIPE still runs _mark_cancelled and emits the cancelled/130 envelope.

Both are pinned by tests that I verified fail against the old arrangement:

• test_a_socket_epipe_is_still_reported_as_a_lost_server — socket-side EPIPE, asserts exit 1 + ws_disconnected + error/server_died. Reverting the gate to isinstance(e, BrokenPipeError) fails it.
• test_ctrl_c_ending_in_epipe_still_reports_cancelled — fails against the exact pre-fix arrangement.
• test_cancellation_outranks_a_closed_stdout — both signals at once; fails if the gate is moved back above the token check.

[github-actions]

🟠 High — The local async branch now sequences the only pre-envelope carrier of prompt_id behind the unguarded jobs_state.write(state) at line 466, which raises ValueError when the server's prompt_id fails state_path's ^[a-zA-Z0-9_\-]{1,128}$ check and OSError on an unwritable state dir or full disk. ValueError isn't in the handler tuple (traceback, no envelope); OSError is misreported as ws_disconnected and, since wait_state is None here, the details omit the prompt_id — so an already-accepted job becomes unaddressable. Use the existing best-effort _write_state helper, which swallows both. Raised by 3 of 8 reviewers (claude-opus-5-thinking-max adversarial, claude-opus-5-thinking-max edge-case, gpt-5.6-sol-max adversarial).

[mattmillerai]

Fixed — routed through _write_state.

You're right that this PR is what makes it matter: before the reordering, queued (the only pre-envelope carrier of the prompt_id) was emitted before the write, so a write failure still left the caller holding the id of an accepted job. Sequencing the emit behind an unguarded jobs_state.write handed that guarantee back. ValueError escaped as a traceback and OSError was misreported as ws_disconnected with the prompt_id omitted (since wait_state is None on this branch), exactly as described.

state_file is now _write_state(state), and the emit reports the outcome as a new nullable state_file field on the event rather than being gated on it — losing the id of an accepted job is the worse failure. See the reply on the _emit_queued thread for that part.

Pinned by test_a_failed_write_still_announces_but_reports_null, parametrized over OSError and ValueError: asserts exit 0, prompt_id still present, state_file: null. Verified it fails when the call is reverted to jobs_state.write.

[github-actions]

🟠 High — Same hazard on both cloud branches: jobs_state.write(state) at lines 984 and 1037 is unguarded and execute_cloud catches neither ValueError (prompt_id rejected by state_path) nor OSError, so a storage failure after the billable submit kills the run with a traceback, no envelope, and no queued line — the caller never learns the prompt_id of a job it is already paying for, which is exactly the loss this reordering is meant to prevent. Route both through a best-effort wrapper like _write_state. Raised by 3 of 8 reviewers (claude-opus-5-thinking-max edge-case, gpt-5.6-sol-max edge-case, kimi-k2.7-code edge-case).

[mattmillerai]

Fixed — both cloud submit-time writes now go through the best-effort _write_state.

Agreed on the framing: this is the loss the reordering exists to prevent, and doing it after a billable submit is the worst place for it. execute_cloud catches neither ValueError (from state_path() rejecting the id) nor OSError, so either killed the run with a traceback and no queued line.

Pinned by test_cloud_survives_a_failed_write_after_the_billable_submit, parametrized over both exception types: asserts exit 0, prompt_id still emitted, state_file: null. Verified it fails when the async-branch call is reverted to jobs_state.write.

[github-actions]

🟠 High — Re-raising BrokenPipeError on the --wait path removes the only code path that finalized the on-disk record, leaving the submit-time record from line 386 at status="running" with watcher_pid=None forever: jobs ls only reaps non-terminal records whose recorded watcher_pid is dead, and --wait never sets one, so the phantom job is unreapable and jobs wait on it burns its full timeout. Consider marking the record (or spawning a detached watcher) before re-raising. Raised by 4 of 8 reviewers (claude-opus-5-thinking-max adversarial, claude-opus-5-thinking-max edge-case, gemini-3.1-pro adversarial, gemini-3.1-pro edge-case).

[mattmillerai]

Fixed — the bail-out now hands the record off before re-raising.

Confirmed the phantom is this PR's doing: pre-PR a BrokenPipeError here was caught by the OSError handler and finalized the record as error/server_died, so re-raising removed the only path that moved it off running. And --wait never sets watcher_pid, so jobs ls's reap (jobs.py:211-222) could never touch it.

Took your first suggestion, since the job genuinely is still running server-side — marking it terminal would be a lie. New _hand_off_record helper: spawn a detached watcher for the prompt, and if that fails, record watcher_pid = os.getpid() so the reaper finishes the job off once this process is gone. Guarded against raising, since it runs while the BrokenPipeError is already unwinding toward the silent exit 0 — replacing it would turn that 0 into a traceback.

The same hand-off is wired into the cloud --wait announcement (see the _emit_queued_cloud thread).

Pinned by test_local_wait_leaves_the_running_record_intact (asserts the watcher spawn) and test_local_wait_names_itself_reapable_when_no_watcher_spawns (asserts the os.getpid() fallback). Both fail when the _hand_off_record call is removed.

[github-actions]

🟡 Medium — Placing _emit_queued_cloud() outside the polling try means an EPIPE here exits 0 while the cloud job keeps running and billing, with the record frozen at queued. Unlike the async branch this path never spawns a watcher, so nothing ever advances or reaps that record and jobs status/wait/ls report a permanently queued job. Raised by 3 of 8 reviewers (claude-opus-5-thinking-max adversarial, gemini-3.1-pro adversarial, gemini-3.1-pro edge-case).

[mattmillerai]

Fixed — same hand-off as the local --wait path.

You're right that this branch is the worst case for it: it spawns no watcher by design, so nothing would ever advance or reap the record, and the job keeps running and billing. The emit stays outside the polling try (so none of those handlers can rewrite state for a stdout failure), but it's now wrapped in its own narrow handler:

try:
    _emit_queued_cloud(state_file)
except OSError:
    _hand_off_record(state, where="cloud", notify=notify)
    raise

OSError rather than BrokenPipeError so an ENOSPC/EIO on a redirected stdout gets the hand-off too; the re-raise is unchanged, so __main__ still turns a broken pipe into the documented silent exit 0 while a genuine lost-output error surfaces.

Pinned by test_cloud_wait_hands_off_the_queued_record, which fails when the hand-off is removed.

[github-actions]

🟡 Medium — This emit now sits inside the try whose handler treats nearly every OSError as a WebSocket failure, and only BrokenPipeError is exempted. An ENOSPC/EIO from a redirected or closed stdout will therefore rewrite a perfectly healthy job record to error/server_died even though the server connection is intact. Raised by 3 of 8 reviewers (gemini-3.1-pro adversarial, gpt-5.6-sol-max edge-case, kimi-k2.7-code edge-case).

[mattmillerai]

Fixed by the same change as the BrokenPipeError gate above.

The narrow exemption was the wrong shape for exactly this reason — the guard now keys on who owns the failing fd, not on which OSError subclass it is. Renderer._write_json_line sets _stream_broken on any OSError from its own machine_stream write, so an ENOSPC/EIO from a redirected or closed stdout is attributed to us and re-raised, instead of rewriting a healthy record to error/server_died.

Because the flag lives in the renderer rather than at this one call site, it also covers a stdout failure during the node events watch_execution streams later in the same try — which the narrow exemption never did.

Covered by test_a_failed_write_flags_it_and_still_raises in tests/comfy_cli/output/test_renderer.py, parametrized over BrokenPipeError and OSError(ENOSPC).

[github-actions]

🟡 Medium — _emit_queued takes no signal of whether the preceding persistence actually succeeded, and every caller emits unconditionally — _write_state is best-effort and returns None when the write was skipped or the state dir was unwritable. That silently falsifies the new docs guarantee that a consumer may run comfy jobs status <prompt_id> the moment it reads the line; gate the emit on the returned path or carry the outcome in the event. Raised by 5 of 8 reviewers (gpt-5.6-sol-max adversarial, gpt-5.6-sol-max edge-case, claude-opus-5-thinking-max both types, gemini-3.1-pro edge-case, kimi-k2.7-code edge-case).

[mattmillerai]

Fixed — the outcome is now carried in the event.

Agreed the docs guarantee was overstated. Of your two options I took carry the outcome rather than gate the emit: gating would mean a failed state write also costs the caller the prompt_id of a job the server has already accepted (and on cloud already billed), which is the larger loss and the one this reordering exists to prevent.

So queued gains a nullable state_file field — the path, or null when _write_state skipped/failed. Also updated:

• comfy_cli/schemas/run_event.json declares it as ["string", "null"], so a consumer validating the stream doesn't reject the very case the field signals.
• docs/json-output.md now says the write is best-effort and tells consumers to key off state_file rather than treat the presence of the queued line as proof the record is there.
• Pretty mode prints not written instead of a bare None where the path used to go.

Pinned by TestQueuedReportsWhetherTheRecordLanded (happy path + OSError/ValueError failures, local and cloud) and test_queued_declares_a_nullable_state_file. Verified they fail when the field is dropped from the emit.

[github-actions]

🟢 Low — test_queue_emits_nothing_and_stashes_the_warnings does not request the ndjson_renderer fixture, so the autouse renderer reset leaves get_renderer() in PRETTY mode where Renderer.event() returns early — the assert "queued" not in out is vacuous and would pass against the old code that still emitted from queue(). Add the ndjson_renderer fixture so the assertion actually pins the behavior. Raised by 1 of 8 reviewers (claude-opus-5-thinking-max edge-case).

[mattmillerai]

Not making this change — I checked empirically and the premise doesn't hold: the assertion is live, not vacuous.

ndjson_renderer is declared @pytest.fixture(autouse=True) (tests/comfy_cli/command/test_run_json.py:44), so it installs a fresh Renderer(mode=OutputMode.NDJSON) for every test in the module whether or not the test names it. reset_renderer_for_testing() runs at teardown, not during the test, so get_renderer() is in NDJSON mode for the duration and Renderer.event() does not return early.

Verified two ways rather than by reading:

1. Probe inside this module's fixture scope — get_renderer().is_stream() and ex.renderer.is_stream() are both True, and an event() call does reach capsys stdout.
2. Mutation test, which is the claim that matters. I restored the old queue() body so it emits queued directly again, and re-ran just this class:

>       assert "queued" not in out, f"queue() must not emit the event itself: {out!r}"
E       AssertionError: queue() must not emit the event itself: '{"schema": "event/1", "type": "queued", "prompt_id": "p", ...}'
tests/comfy_cli/command/test_run_json.py:2265: AssertionError
1 failed, 1 passed

So the test does fail against "the old code that still emitted from queue()" — exactly the regression it was written to catch. Adding an explicit ndjson_renderer parameter would be a no-op (same fixture instance either way), so I've left it as is.

Thanks for the pass — the other eight findings in this panel were real and are fixed.

[github-actions]

⚪ Nit — The event stream is now sequenced behind jobs_state.write, which takes locking.file_lock with no timeout (fcntl.flock(fd, LOCK_EX) blocks indefinitely). A stuck holder of <prompt_id>.lock — a SIGSTOPped watcher, a hung NFS mount — now stalls the queued announcement itself, so a consumer blocking on that line hangs unbounded; passing a timeout= would keep the announcement live. Raised by 1 of 8 reviewers (claude-opus-5-thinking-max adversarial).

[mattmillerai]

Real, but deferring to a follow-up ticket rather than fixing here.

Confirmed the mechanics: jobs_state.write (comfy_cli/jobs_state.py:125) takes locking.file_lock(path.with_suffix(".lock")) with no timeout, and locking.file_lock (comfy_cli/locking.py:49) blocks indefinitely when timeout is None. So a stuck holder does stall the announcement, and a consumer blocking on that line hangs unbounded.

Out of scope for this PR because the lock has always been unbounded — this PR only changed which line is sequenced behind the write (the async path already wrote state before emitting its envelope). Fixing it means threading a timeout through a primitive shared by the foreground run paths, the detached watcher (command/job_watcher.py), and jobs ls's stale-watcher reap (command/jobs.py:211-222), and deciding per call site what a timeout means. The run command's submit-time writes now go through the best-effort _write_state, so they can degrade to state_file: null cleanly — but the watcher and the reap write assume the write lands, and making them fail silently on timeout would quietly stop jobs ls reaping. That needs a decision before coding, so it isn't something to bolt on here.

Recorded as a follow-up (low severity) with that analysis and the proposed write(state, *, lock_timeout=…) shape. Worth noting file_lock's own docstring already flags that timeout is best-effort and that flock on NFS is often a silent no-op, so the hung-NFS half may need a different primitive rather than a timeout.

[mattmillerai]

🤖 The reviews loop filed Linear follow-up ticket(s) for review thread(s) deferred as out of scope for this PR:

• BE-6288 — Give jobs_state.write a lock acquisition timeout so a stuck holder can't hang the CLI forever — filed as agent-spike (premise unverified)

The following carry agent-spike instead of agent-ok because their reachability claim was not backed by evidence (BE-5378) — the claim is investigated before any code is written, and "the premise does not hold" is a valid, successful outcome:

• Give jobs_state.write a lock acquisition timeout so a stuck holder can't hang the CLI forever — no reachability block in the proposal