diff --git a/README.md b/README.md index 96fd1b1..a626e45 100644 --- a/README.md +++ b/README.md @@ -101,6 +101,7 @@ The list above is the shortest accurate summary. If you want the execution path - If another Modal prompt is already active when a new prompt is submitted, queue-time setup updates for the new prompt are held back so its nodes do not repaint until ComfyUI actually starts executing it. - Completed node progress fades out briefly instead of disappearing abruptly or remaining stuck at 100%. - Completed remote nodes use a dark blue outline and fill while later remote nodes continue executing; during finalizing, completed nodes switch to a darker blue outline with pulsing fill while Modal outputs return locally. + - When a remote node starts executing, the local UI also marks its reported upstream remote ancestors complete, so quick or progress-less upstream nodes do not stay purple just because they never emitted their own completion update. - The global status pill stays in the orange waiting state while Modal capacity is still starting up and only switches to running after streamed remote execution identifies an active node or emits real node progress. - Prompt errors clear prompt-scoped remote node visual state immediately, so stale setup/ready colors do not leak into a resubmitted prompt. - Prompt success, failure, and interruption mark that prompt terminal in the frontend so late remote progress cannot resurrect stale node overlays or the global status pill. diff --git a/docs/LEARNINGS.md b/docs/LEARNINGS.md index 4efe37c..ba63117 100644 --- a/docs/LEARNINGS.md +++ b/docs/LEARNINGS.md @@ -233,3 +233,4 @@ - Local proxy execution is not proof that a Modal worker is running the graph. The frontend should keep the global pill in an orange waiting state while ComfyUI is executing the proxy and Modal is still provisioning capacity, and only switch to green running once streamed remote feedback includes an active remote node or actual node-progress data. - A Modal cancellation flag can be written before the remote method body starts if the user cancels while Modal is still provisioning capacity. Remote execution registration must not clear that key on entry, and the local proxy still needs a bounded post-interrupt grace period so ComfyUI is released even when remote code is slow to observe the interrupt. - A split mapped proxy can carry an embedded `static_phase` that is executed later for mapped lane seeding. Any original boundary inputs required by that embedded phase must still be wired onto the mapped proxy node, even if they are not part of the mapped payload's top-level `boundary_inputs`; otherwise the remote worker reaches lane seeding with no hydrated value for names like `remote_input_0`. +- Remote execution streams do not reliably produce explicit completion events for every node, especially quick nodes or nodes without numeric progress. The local stream bridge still has the remote subgraph prompt, so it should send dependency ancestors for each active/progress node and let the frontend mark those ancestors complete before painting the current node active. diff --git a/remote/modal_app.py b/remote/modal_app.py index 66a263d..1eb915d 100644 --- a/remote/modal_app.py +++ b/remote/modal_app.py @@ -2079,6 +2079,7 @@ def _emit_local_modal_status( status_message: str | None = None, status_current: int | None = None, status_total: int | None = None, + completed_ancestor_node_ids: list[str] | None = None, ) -> None: """Forward remote execution progress into the local ComfyUI websocket stream.""" if client_id is None: @@ -2107,6 +2108,8 @@ def _emit_local_modal_status( payload["status_current"] = int(status_current) if status_total is not None: payload["status_total"] = int(status_total) + if completed_ancestor_node_ids: + payload["completed_ancestor_node_ids"] = list(completed_ancestor_node_ids) prompt_server.send_sync("modal_status", payload, client_id) @@ -2125,6 +2128,7 @@ def _emit_local_modal_progress( aggregate_only: bool = False, setup_only: bool = False, cached_hit: bool = False, + completed_ancestor_node_ids: list[str] | None = None, ) -> None: """Forward remote numeric node progress into the local ComfyUI websocket stream.""" if client_id is None: @@ -2156,6 +2160,8 @@ def _emit_local_modal_progress( payload["setup_only"] = True if cached_hit: payload["cached_hit"] = True + if completed_ancestor_node_ids: + payload["completed_ancestor_node_ids"] = list(completed_ancestor_node_ids) prompt_server.send_sync("modal_progress", payload, client_id) @@ -2350,6 +2356,37 @@ def _progress_stream_event_metadata(stream_event: dict[str, Any]) -> dict[str, s } +def _remote_prompt_ancestor_node_ids(payload: dict[str, Any], node_id: str | None) -> list[str]: + """Return subgraph dependency ancestors for one currently executing remote node.""" + if node_id is None: + return [] + prompt = payload.get("subgraph_prompt", {}) + if not isinstance(prompt, dict) or node_id not in prompt: + return [] + + ancestors: set[str] = set() + pending = [node_id] + while pending: + current_node_id = str(pending.pop()) + prompt_node = prompt.get(current_node_id) + if not isinstance(prompt_node, dict): + continue + inputs = prompt_node.get("inputs") or {} + if not isinstance(inputs, dict): + continue + for input_value in inputs.values(): + if not _is_link(input_value): + continue + upstream_node_id = str(input_value[0]) + if upstream_node_id == node_id or upstream_node_id in ancestors: + continue + if upstream_node_id not in prompt: + continue + ancestors.add(upstream_node_id) + pending.append(upstream_node_id) + return sorted(ancestors) + + def _should_stream_remote_progress(payload: dict[str, Any]) -> bool: """Return whether the local client has enough context to mirror remote node progress.""" extra_data = payload.get("extra_data") or {} @@ -2608,6 +2645,10 @@ def _consume_remote_payload_stream( progress_metadata["real_node_id"] if progress_metadata is not None else None ) progress_node_id = real_node_id or display_node_id + completed_ancestor_node_ids = _remote_prompt_ancestor_node_ids( + payload, + progress_node_id, + ) if lane_id is not None: _remember_mapped_lane_node_id(payload, lane_id, progress_node_id) elif suppress_status_stream and not aggregate_only and progress_metadata is not None: @@ -2625,17 +2666,17 @@ def _consume_remote_payload_stream( stream_event.get("max"), lane_id, ) - _emit_local_modal_progress( - prompt_id=prompt_id, - client_id=client_id, - node_id=str(reported_node_id), - value=float(stream_event.get("value", 0.0)), - max_value=float(stream_event.get("max", 1.0)), - display_node_id=display_node_id, - real_node_id=real_node_id, - lane_id=lane_id, - clear=bool(stream_event.get("clear", False)), - item_index=( + progress_kwargs = { + "prompt_id": prompt_id, + "client_id": client_id, + "node_id": str(reported_node_id), + "value": float(stream_event.get("value", 0.0)), + "max_value": float(stream_event.get("max", 1.0)), + "display_node_id": display_node_id, + "real_node_id": real_node_id, + "lane_id": lane_id, + "clear": bool(stream_event.get("clear", False)), + "item_index": ( int(stream_event["item_index"]) if stream_event.get("item_index") is not None else ( @@ -2644,8 +2685,13 @@ def _consume_remote_payload_stream( else None ) ), - aggregate_only=aggregate_only, - ) + "aggregate_only": aggregate_only, + } + if completed_ancestor_node_ids: + progress_kwargs["completed_ancestor_node_ids"] = ( + completed_ancestor_node_ids + ) + _emit_local_modal_progress(**progress_kwargs) continue if event_type == "executed": reported_node_id = stream_event.get("node_id") @@ -2825,6 +2871,12 @@ def _consume_remote_payload_stream( if stream_event.get("active_node_id") is not None else None ), + completed_ancestor_node_ids=_remote_prompt_ancestor_node_ids( + payload, + str(stream_event["active_node_id"]) + if stream_event.get("active_node_id") is not None + else None, + ) or None, active_node_class_type=( str(stream_event["active_node_class_type"]) if stream_event.get("active_node_class_type") is not None diff --git a/tests/test_modal_executor_node.py b/tests/test_modal_executor_node.py index 8be8003..bff0d1a 100644 --- a/tests/test_modal_executor_node.py +++ b/tests/test_modal_executor_node.py @@ -7947,6 +7947,54 @@ def test_consume_remote_payload_stream_keeps_static_execute_node_progress_when_s ] +def test_consume_remote_payload_stream_marks_progress_node_ancestors_complete( + remote_modal_app_module: Any, + serialization_module: Any, + monkeypatch: Any, +) -> None: + """Streamed progress should tell the UI which upstream remote nodes can be completed.""" + progress_calls: list[dict[str, Any]] = [] + + monkeypatch.setattr( + remote_modal_app_module, + "_emit_local_modal_progress", + lambda **kwargs: progress_calls.append(kwargs), + ) + + payload = { + "component_id": "3", + "prompt_id": "prompt-1", + "component_node_ids": ["1", "2", "3"], + "extra_data": {"client_id": "client-1"}, + "subgraph_prompt": { + "1": {"class_type": "Loader", "inputs": {}}, + "2": {"class_type": "Condition", "inputs": {"model": ["1", 0]}}, + "3": {"class_type": "Sampler", "inputs": {"model": ["1", 0], "conditioning": ["2", 0]}}, + }, + } + stream_events = iter( + [ + { + "kind": "progress", + "event_type": "node_progress", + "node_id": "3", + "value": 1.0, + "max": 4.0, + }, + { + "kind": "result", + "outputs": serialization_module.serialize_node_outputs(("done",)), + }, + ] + ) + + response = remote_modal_app_module._consume_remote_payload_stream(payload, stream_events) + + assert serialization_module.deserialize_node_outputs(response) == ("done",) + assert progress_calls[0]["node_id"] == "3" + assert progress_calls[0]["completed_ancestor_node_ids"] == ["1", "2"] + + def test_consume_remote_payload_stream_clears_static_execute_node_progress_on_suppressed_completion( remote_modal_app_module: Any, serialization_module: Any, diff --git a/tests/test_web_modal_toggle.py b/tests/test_web_modal_toggle.py index 70c3343..3870c34 100644 --- a/tests/test_web_modal_toggle.py +++ b/tests/test_web_modal_toggle.py @@ -223,6 +223,17 @@ def test_streamed_modal_node_progress_fades_previous_active_node() -> None: assert "setPromptActiveNode(promptId, progressNodeId);" in source +def test_streamed_modal_node_progress_completes_reported_ancestors() -> None: + """When a remote node becomes active, reported upstream nodes should be marked complete.""" + source = _modal_toggle_source() + + assert "function completeRemoteAncestorsBeforeActiveNode(promptId, ancestorNodeIds, activeNodeId)" in source + assert "detail.completed_ancestor_node_ids" in source + assert "setNodesPhase(completedNodeIds, STATE_COMPLETE, promptId);" in source + assert "completeRemoteAncestorsBeforeActiveNode(\n promptId,\n detail.completed_ancestor_node_ids,\n progressNodeId,\n );" in source + assert "completeRemoteAncestorsBeforeActiveNode(\n promptId,\n detail.completed_ancestor_node_ids,\n nextActiveNodeId,\n );" in source + + def test_mapped_parallel_modal_progress_renders_multiple_lane_bars() -> None: """Parallel mapped Modal runs should render one local progress lane per active worker.""" source = _modal_toggle_source() diff --git a/web/modal_toggle.js b/web/modal_toggle.js index f63a5b3..e518008 100644 --- a/web/modal_toggle.js +++ b/web/modal_toggle.js @@ -902,6 +902,32 @@ function setPromptActiveNode(promptId, activeNodeId) { promptState.activeNodeId = activeNodeId ? String(activeNodeId) : null; } +/** + * Mark known upstream remote nodes complete when a downstream node is executing. + * @param {string} promptId + * @param {any[] | undefined} ancestorNodeIds + * @param {string | null} activeNodeId + */ +function completeRemoteAncestorsBeforeActiveNode(promptId, ancestorNodeIds, activeNodeId) { + if (!Array.isArray(ancestorNodeIds) || ancestorNodeIds.length === 0) { + return; + } + const completedNodeIds = Array.from( + new Set( + ancestorNodeIds + .map((nodeIdValue) => String(nodeIdValue)) + .filter((nodeIdValue) => nodeIdValue && nodeIdValue !== activeNodeId), + ), + ); + if (completedNodeIds.length === 0) { + return; + } + for (const completedNodeId of completedNodeIds) { + fadeNodeProgress(completedNodeId, promptId); + } + setNodesPhase(completedNodeIds, STATE_COMPLETE, promptId); +} + /** * Return the numeric progress payload for one node when it belongs to the prompt. * @param {string} nodeIdValue @@ -2137,6 +2163,11 @@ function handleModalStatus(event) { fadeNodeProgress(previousActiveNodeId, promptId); setNodesPhase([previousActiveNodeId], STATE_COMPLETE, promptId); } + completeRemoteAncestorsBeforeActiveNode( + promptId, + detail.completed_ancestor_node_ids, + nextActiveNodeId, + ); if (nextActiveNodeId) { setNodesPhase([nextActiveNodeId], STATE_ACTIVE, promptId); } @@ -2264,6 +2295,11 @@ function handleModalProgress(event) { fadeNodeProgress(previousActiveNodeId, promptId); setNodesPhase([previousActiveNodeId], STATE_COMPLETE, promptId); } + completeRemoteAncestorsBeforeActiveNode( + promptId, + detail.completed_ancestor_node_ids, + progressNodeId, + ); setPromptActiveNode(promptId, progressNodeId); setNodesPhase([progressNodeId], STATE_ACTIVE, promptId); setNodeProgress(