diff --git a/README.md b/README.md index cd425c3..39302e0 100644 --- a/README.md +++ b/README.md @@ -179,6 +179,7 @@ Remote runtime behavior: - Mapped progress still exposes per-item lane metadata and aggregate completion, but it is now emitted by the single mapped runtime rather than by a local fan-out queue of separate remote calls. - The remote worker only forwards numeric node progress for meaningful progress states. Trivial `0/1` updates from ordinary upstream nodes are ignored so sampler-style progress bars stay attached to nodes like `KSampler` instead of appearing on loaders or text encoders. - Static sub-runs inside hybrid mapped execution now also forward their lane-less execute-node progress even when coarse per-phase status chatter is suppressed, so one-time upstream samplers still show real progress bars while mapped item runs keep their quieter stream behavior. +- Those suppressed static progress events now emit an explicit clear when the static sub-run finishes, so one-time upstream sampler bars disappear promptly at completion instead of sticking at `100%` until the whole prompt ends. - Mapped remote components can now contain both one-time execute targets and per-item execute targets. A common case is two remote `KSampler` nodes sharing one upstream `Load Diffusion Model`, where only the sampler fed by `ModalMapInput` should fan out per latent while the sibling sampler still runs exactly once. - Queue-time rewrite now materializes explicit static-to-mapped boundaries for invariant upstream outputs. That lets the mapped phase consume only the values it really depends on instead of carrying the whole coarse component prompt back through every item run. - Hybrid mapped sub-runs now trim their prompt down to the dependency closure of the specific `execute_node_ids` they are about to run. That keeps a static-only sub-run from validating or executing the unrelated mapped branch just because both branches originally lived in the same coarse remote component. diff --git a/docs/LEARNINGS.md b/docs/LEARNINGS.md index 18ccfb5..191ba53 100644 --- a/docs/LEARNINGS.md +++ b/docs/LEARNINGS.md @@ -132,3 +132,4 @@ - Restoring persisted outputs before `PromptExecutor.execute()` is too early. `execute()` immediately rebuilds the live cache-key set with its own `set_prompt()` call, so a distributed cache restore must run inside or immediately after that phase if the restored entries are supposed to participate in the actual execution walk. - Mapped progress lanes have to preserve the scheduler-assigned lane id all the way through the remote worker. If the worker hardcodes every item's node progress to lane `0`, concurrent mapped items overwrite each other in the frontend and upstream sampler bars appear to flicker or disappear entirely. - Static mapped sub-runs cannot reuse the same "suppress status chatter" rule as per-item mapped lanes for numeric node progress. Those static phases usually do not have a lane id at all, so the local relay has to keep lane-less `node_progress` events for execute/boundary nodes or upstream samplers like `KSampler` will run remotely without ever showing a bar in the UI. +- Once those lane-less static progress events are forwarded, they also need their own explicit clear path. Suppressed static streams do not forward the terminal `execution_success` status event into the frontend, so relying on phase completion to clear ordinary progress leaves the last `100%` sampler bar stuck onscreen. diff --git a/remote/modal_app.py b/remote/modal_app.py index 9e0fefc..5f6e2f8 100644 --- a/remote/modal_app.py +++ b/remote/modal_app.py @@ -1093,6 +1093,34 @@ def _progress_stream_event_node_id(stream_event: dict[str, Any]) -> str | None: return None +def _progress_stream_event_metadata(stream_event: dict[str, Any]) -> dict[str, str | None] | None: + """Return normalized metadata for one streamed progress event.""" + reported_node_id = ( + str(stream_event["node_id"]) + if stream_event.get("node_id") is not None + else None + ) + display_node_id = ( + str(stream_event["display_node_id"]) + if stream_event.get("display_node_id") is not None + else reported_node_id + ) + real_node_id = ( + str(stream_event["real_node_id"]) + if stream_event.get("real_node_id") is not None + else None + ) + filter_node_id = real_node_id or reported_node_id or display_node_id + if filter_node_id is None: + return None + return { + "node_id": reported_node_id or filter_node_id, + "display_node_id": display_node_id, + "real_node_id": real_node_id, + "filter_node_id": filter_node_id, + } + + 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 {} @@ -1277,13 +1305,17 @@ def _consume_remote_payload_stream( node_ids = [str(node_id) for node_id in payload.get("component_node_ids", [])] suppress_status_stream = bool(payload.get("suppress_status_stream")) result_payload: bytes | bytearray | None = None + suppressed_progress_node_metadata: dict[str, dict[str, str | None]] = {} for stream_event in stream_events: event_kind = str(stream_event.get("kind", "")) if event_kind == "progress": event_type = str(stream_event.get("event_type", "")) if event_type == "node_progress": - filter_node_id = _progress_stream_event_node_id(stream_event) + progress_metadata = _progress_stream_event_metadata(stream_event) + filter_node_id = ( + progress_metadata["filter_node_id"] if progress_metadata is not None else None + ) lane_id = ( str(stream_event["lane_id"]) if stream_event.get("lane_id") is not None @@ -1307,23 +1339,27 @@ def _consume_remote_payload_stream( stream_event.get("real_node_id"), ) continue - reported_node_id = stream_event.get("node_id") - if reported_node_id is None: - reported_node_id = filter_node_id + reported_node_id = ( + progress_metadata["node_id"] if progress_metadata is not None else None + ) if reported_node_id is not None: display_node_id = ( - str(stream_event["display_node_id"]) - if stream_event.get("display_node_id") is not None + progress_metadata["display_node_id"] + if progress_metadata is not None else str(reported_node_id) ) real_node_id = ( - str(stream_event["real_node_id"]) - if stream_event.get("real_node_id") is not None - else None + progress_metadata["real_node_id"] if progress_metadata is not None else None ) progress_node_id = real_node_id or display_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: + suppressed_progress_node_metadata[str(progress_metadata["filter_node_id"])] = { + "node_id": str(reported_node_id), + "display_node_id": display_node_id, + "real_node_id": real_node_id, + } logger.debug( "Forwarding streamed Modal node progress for component=%s node_id=%s real_node_id=%s value=%s max=%s lane_id=%s.", payload.get("component_id"), @@ -1453,6 +1489,20 @@ def _consume_remote_payload_stream( stream_event.get("active_node_id"), ) if suppress_status_stream: + remote_phase = str(stream_event.get("phase", "executing")) + if remote_phase in {"execution_success", "execution_error", "execution_interrupted"}: + for progress_metadata in suppressed_progress_node_metadata.values(): + _emit_local_modal_progress( + prompt_id=prompt_id, + client_id=client_id, + node_id=str(progress_metadata["node_id"]), + value=0.0, + max_value=1.0, + display_node_id=progress_metadata["display_node_id"], + real_node_id=progress_metadata["real_node_id"], + clear=True, + ) + suppressed_progress_node_metadata.clear() continue remote_phase = str(stream_event.get("phase", "executing")) if remote_phase == "execution_success": diff --git a/tests/test_modal_executor_node.py b/tests/test_modal_executor_node.py index c2c8f9a..182a1c0 100644 --- a/tests/test_modal_executor_node.py +++ b/tests/test_modal_executor_node.py @@ -3214,6 +3214,83 @@ def test_consume_remote_payload_stream_keeps_static_execute_node_progress_when_s ] +def test_consume_remote_payload_stream_clears_static_execute_node_progress_on_suppressed_completion( + remote_modal_app_module: Any, + serialization_module: Any, + monkeypatch: Any, +) -> None: + """Suppressed static sub-runs should emit an explicit clear for lane-less node progress on completion.""" + 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": "1::static", + "prompt_id": "prompt-1", + "component_node_ids": ["1", "2", "12", "4"], + "execute_node_ids": ["12", "4"], + "boundary_outputs": [ + {"node_id": "4", "output_index": 0, "io_type": "LATENT", "is_list": False} + ], + "extra_data": {"client_id": "client-1"}, + "suppress_status_stream": True, + } + stream_events = iter( + [ + { + "kind": "progress", + "event_type": "node_progress", + "node_id": "1", + "display_node_id": "1", + "real_node_id": "12", + "value": 5.0, + "max": 20.0, + }, + { + "kind": "progress", + "phase": "execution_success", + }, + { + "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 == [ + { + "prompt_id": "prompt-1", + "client_id": "client-1", + "node_id": "1", + "value": 5.0, + "max_value": 20.0, + "display_node_id": "1", + "real_node_id": "12", + "lane_id": None, + "clear": False, + "item_index": None, + "aggregate_only": False, + }, + { + "prompt_id": "prompt-1", + "client_id": "client-1", + "node_id": "1", + "value": 0.0, + "max_value": 1.0, + "display_node_id": "1", + "real_node_id": "12", + "clear": True, + }, + ] + + def test_consume_remote_payload_stream_filters_static_sibling_ui_events_from_mapped_items( 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 c7f138f..f75b178 100644 --- a/tests/test_web_modal_toggle.py +++ b/tests/test_web_modal_toggle.py @@ -153,6 +153,8 @@ def test_mapped_parallel_modal_progress_renders_multiple_lane_bars() -> None: assert "const laneColors = [" in source assert 'const badgeText = hasBatchBadge' in source assert 'const badgeY = panelY + panelPaddingY;' in source + assert "if (detail.clear) {" in source + assert "clearNodeProgress(progressNodeId, promptId);" in source assert 'let laneY = barY;' in source assert "laneNodeIdsByLane: new Map()," in source assert "function deleteNodeProgressLane(nodeIdValue, promptId, laneId)" in source diff --git a/web/modal_toggle.js b/web/modal_toggle.js index 9cfe9c2..22288c3 100644 --- a/web/modal_toggle.js +++ b/web/modal_toggle.js @@ -1613,6 +1613,10 @@ function handleModalProgress(event) { ); return; } + if (detail.clear) { + clearNodeProgress(progressNodeId, promptId); + return; + } if (readyNodeIds.length > 0) { setNodesPhase(readyNodeIds, STATE_READY, promptId); }