Align If execution with iterator frames

invokeai/app/services/shared/README.md

@@ -837,9 +837,15 @@ Run `C` -> `D:0` -> enqueue `D`. Run `D` -> done.
     (a node between the `Collect` and the `Iterate`, or extra collected items) can change the length, and selecting
     the upstream axis over its reopening pairs it with per-group frames; both would drop or invent frames.
   - A branch whose open axes, or the axes a branch `Collect` keeps, are deeper than a per-item condition's, unless
-    both branches carry the same deeper axes.
-  - A root-frame condition choosing between branches with different deeper axes. When both branches share nested
-    deeper axes, a root-frame condition runs the `If` once per frame of the innermost one.
+    both branches carry the same deeper axes. Each `Collect.item` closes its innermost axis and keeps the outer axes;
+    this identity is followed through chained `Collect`s and ordinary passthrough nodes. Separate branch Collect chains
+    may retain the same axis identities and frames; sharing a Collect node is not required. Matching lengths alone do
+    not establish alignment. An asymmetric or distinct-axis Collect chain is rejected before invocation. A downstream
+    `Collect` retains empty outer groups.
+  - A root-frame condition choosing between branches with different deeper axes. When both branches retain the same
+    nested axes, a root-frame condition runs the `If` once per available frame of the innermost axis. Selecting a root
+    value over an axis reached only through an unselected `For` final output is the root-frame exception described
+    above; literal and computed root conditions have the same result cardinality.
   Selected-branch frame cardinality is not supported.
 - A completed `Collect` on a branch closes its item axis; the `If` does not wait for an `Iterate` that only that
   `Collect` consumes. When the `Collect` keeps an outer frame open and is selected, the `If` runs once per preserved
@@ -849,17 +855,17 @@ Run `C` -> `D:0` -> enqueue `D`. Run `D` -> done.
   through the loop linkage) is gated with that branch: for a frame that rejects the branch, neither the `For` nor its
   body runs. A gated `For` brings in a `For` nested in its body together with that body. Two `For`s stay ungated. A
   `For` nested in the body of an ungated `For` runs with that loop. A `For` whose final output feeds the `If` directly
-  is attached only after it finalizes, and the `If` is not prepared before that, so gating it on the `If` would
-  deadlock. Known limitation: with a per-item condition, such a `For` and its body run even for items that reject the
-  branch, and an error they raise is recorded.
+  attaches after it finalizes; with a per-item condition, rejected frames do not run the `For` or its body, and selected
+  frames wait for its final output before the `If` input is attached.
 - An `If` has no execution in a frame where the innermost iterator in its ancestry has no items. When an `Iterate` in
   its iterator ancestry, whether it feeds `If.condition` or a branch, has no items at all, no `If` execution is created,
   the branch sources are retired empty, the downstream `Collect` produces an empty collection, and the state completes
   without an error. The same holds for a literal condition whose selected branch runs under an empty `Iterate` reached
   only through a `For` final output. Retirement may add the branch source IDs to `executed_history`; neither branch
-  invocation runs and no branch result is created. Known limitation: with a wired condition, an `If` whose branch
-  reaches an empty `Iterate` only through a `For` final output stalls, whether the condition is root-frame or per item
-  of that `Iterate`.
+  invocation runs and no branch result is created. With a wired root-frame condition and one parent `Iterate` carried
+  through a selected `For` final output, the `If` broadcasts over the parent frames; an empty parent `Iterate` gives it
+  no executions and the downstream `Collect` remains empty. Multiple unrelated open frames that cannot be paired still
+  fail validation before execution.
 - The SQLite queue/processor path has evidence for cancellation before and after `If` resolution and retry from each
   boundary for both condition polarities. A canceled attempt keeps its activation ledger and cannot resume; retry
   starts a fresh state and must emit a fresh matching activation token before selected-only completion.

invokeai/app/services/shared/graph_if_activation.py

@@ -9,11 +9,11 @@
 
 
 class _IfActivationController:
-    """Own activation admission for concrete execution nodes.
+    """Compile activation dependencies for source nodes on compatibility paths.
 
     Unlike the legacy compiler, this controller does not lower or cache a
-    branch projection. It answers one readiness question for one prepared
-    execution node from the current source graph and durable execution frame.
+    branch projection. It derives frame-scoped branch requirements from the
+    current source graph for the fallback scheduler.
     """
 
     def __init__(self, state: "GraphExecutionState") -> None:
@@ -45,8 +45,8 @@ def _branch_sources(self, if_node_id: str, branch_field: str, source_graph: "nx.
         # A For's loop body feeds its ForReturn, which reaches the For only through the loop linkage. The body belongs
         # to the branch whenever its For does, and leaves it with the For. Fors are visited outermost first, so a gated
         # For brings in a For nested in its body together with that body. Two Fors stay ungated. One whose final
-        # output enters the If directly is attached only after it finalizes, and the If is not prepared before that,
-        # so the For cannot wait for the If's verdict. One nested in an ungated For's body is expanded by that loop,
+        # output enters the If directly is attached after it finalizes. With a wired condition the For and its body
+        # can still be gated once that condition resolves. One nested in an ungated For's body is expanded by that loop,
         # which needs it to run. (A For enclosing the If is a frame iterator of the If and was removed above.)
         for_by_return_id: dict[str, str] = {}
         if any(isinstance(self._state.graph.nodes.get(node_id), ForInvocation) for node_id in branch_sources):
@@ -61,6 +61,7 @@ def _branch_sources(self, if_node_id: str, branch_field: str, source_graph: "nx.
                 if (
                     for_id not in branch_sources
                     or for_id in direct_sources
+                    and branch_field == self._get_source_level_selected_field(if_node_id)
                     or any(
                         for_id in body_ids_by_for[enclosing_id] and enclosing_id not in for_by_return_id.values()
                         for enclosing_id in body_ids_by_for
@@ -127,14 +128,6 @@ def _branch_memberships(self, source_node_id: str) -> tuple[tuple[str, tuple[str
         self._branch_memberships_by_source[source_node_id] = memberships
         return memberships
 
-    def get_dependencies(self, exec_node_id: str) -> tuple[ActivationDependency, ...]:
-        """Return activation requirements for one concrete prepared node."""
-
-        source_node_id = self._state._prepared_registry().get_source_node_id(exec_node_id)
-        if not self._get_if_node_ids():
-            return ()
-        return self.get_source_dependencies(source_node_id, self._state._get_iteration_path(exec_node_id))
-
     def get_source_dependencies(
         self, source_node_id: str, iteration_path: tuple[int, ...] = ()
     ) -> tuple[ActivationDependency, ...]:
@@ -145,43 +138,3 @@ def get_source_dependencies(
             for if_node_id, fields in self._branch_memberships(source_node_id)
             if len(fields) == 1
         )
-
-    def is_source_admitted(self, source_node_id: str, iteration_path: tuple[int, ...] = ()) -> bool:
-        dependencies = self.get_source_dependencies(source_node_id, iteration_path)
-        return bool(
-            not dependencies
-            or all(self._state._is_activation_dependency_satisfied(dependency) for dependency in dependencies)
-        )
-
-    def is_source_rejected(self, source_node_id: str, iteration_path: tuple[int, ...] = ()) -> bool:
-        return any(
-            self._state._is_activation_dependency_rejected(dependency)
-            for dependency in self.get_source_dependencies(source_node_id, iteration_path)
-        )
-
-    def is_source_inactive(self, source_node_id: str, iteration_path: tuple[int, ...] = ()) -> bool:
-        """Return whether a source belongs exclusively to a rejected If branch."""
-
-        if iteration_path:
-            return self.is_source_rejected(source_node_id, iteration_path)
-
-        relevant_if_ids = {if_node_id for if_node_id, _ in self._branch_memberships(source_node_id)}
-        if not relevant_if_ids:
-            return False
-
-        frames: set[tuple[int, ...]] = set()
-        for if_node_id in relevant_if_ids:
-            frames.update(
-                self._state._get_iteration_path(exec_node_id)
-                for exec_node_id in self._state._prepared_registry().get_prepared_ids(if_node_id)
-            )
-        if not frames:
-            return False
-
-        for frame in frames:
-            dependencies = self.get_source_dependencies(source_node_id, frame)
-            if dependencies and not any(
-                self._state._is_activation_dependency_rejected(dependency) for dependency in dependencies
-            ):
-                return False
-        return True