fix: clear suppressed static progress bars

This commit is contained in:
Ken Simpson
2026-04-19 22:00:34 -07:00
parent 21295dd45e
commit fd3787d9fe
6 changed files with 144 additions and 9 deletions
+1
View File
@@ -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.
+1
View File
@@ -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.
+59 -9
View File
@@ -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":
+77
View File
@@ -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,
+2
View File
@@ -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
+4
View File
@@ -1613,6 +1613,10 @@ function handleModalProgress(event) {
);
return;
}
if (detail.clear) {
clearNodeProgress(progressNodeId, promptId);
return;
}
if (readyNodeIds.length > 0) {
setNodesPhase(readyNodeIds, STATE_READY, promptId);
}