Fix nested collector iteration scope

[JPPhoto]

Summary

Fixes nested collector scoping in graph execution.

Collectors nested under iterators now group prepared parent exec nodes by iteration path. A collector collapses the immediate iterator feeding its item input while preserving enclosing iterator paths, preventing collections from unrelated outer iterations from being mixed before downstream consumers run.

This fixes workflows where an outer item is decomposed into a per-item collection, that collection is iterated, and the results are collected again. The observed case was an XYImage tiling workflow where tiles from separate source images could be combined into the same collected input.

Also updates the execution engine README to document collector grouping, preserved iteration paths, runtime rehydration, and ready queue ordering.

Related Issues / Discussions

N/A

QA Instructions

I verified this fix against current main (9152232189c02d0a3807e18af682e0fc2bccaffa) by applying only the updated tests/test_graph_execution_state.py to a temporary main worktree and running the new tests.

Key failures:

test_graph_collector_nested_under_outer_iterator_collects_only_current_outer_iteration_items

Expected:
[[0, 1], [10, 11]]

Actual on main:
[[0, 1, 10, 11]]

test_graph_collector_reuses_outer_collection_input_for_each_nested_iterator_group

Expected:
[[100, 0, 1], [100, 10, 11]]

Actual on main:
[[100, 0, 1, 10, 11]]

The test_graph_collector_nested_under_three_iterators_preserves_outer_iteration_paths test also fails in similar fashion.

The helper-level test also fails on main because _get_collect_iteration_mapping_groups() does not exist before this branch.

Manual QA, if the XYGrid community node pack is available:

1. Create a workflow with this shape:

Image Collection (2 visually distinct larger images)
  -> Iterate
  -> Image To XYImage Tiles
  -> Iterate
  -> XYImage Expand
  -> image-processing or pass-through step
  -> XYImage Collect
  -> Collect
  -> XYImage Tiles To Image
  -> Save Image

2. Configure tiling so each source image produces multiple tiles.

3. Run the workflow before and after this fix.

Before this fix:

• The final Collect can mix XYImage Collect outputs from different source images.
• XYImage Tiles To Image may receive tiles from more than one source image at once.
• The output can show cross-image composition, for example one source image's tile(s) combined with another's.

After this fix:

• Each outer source image gets its own collected tile set.
• XYImage Tiles To Image runs separately per source image.
• Outputs reconstruct each source image independently, with no cross-image tile mixing.

Note: I checked simpler built-in-only primitive and built-in tile workflows, and they did not reproduce the pre-fix issue. The manual repro needs a workflow shape where a node under the outer iterator produces a fresh per-outer-item collection that is then iterated and collected downstream.

Merge Plan

Checklist

• The PR has a short but descriptive title, suitable for a changelog
• Tests added / updated (if applicable)
• ❗Changes to a redux slice have a corresponding migration
• Documentation added / updated (if applicable)
• Updated What's New copy (if doing a release after this PR)

[Pfannkuchensack]

Findings

1. Medium - invokeai/app/services/shared/graph.py:414-460

_get_collect_iteration_group_key / _get_collect_iteration_mapping_groups: item inputs
that collapse to an ancestor scope leak across all sibling outer-iteration groups,
re-introducing the exact cross-iteration mixing this PR is meant to prevent.

Evidence chain:

1. _get_collect_iteration_group_key (line 414) computes an item edge's group key as
   path[:-1], dropping the immediate iterator index. An item source sitting directly
   under the outer iterator has path (o,), so it collapses to () - losing which outer
   iteration it belonged to.
2. final_group_keys (lines 440-447) drops any key that is a strict prefix of another, and
   the merge step (lines 449-460) folds every prefix group's mappings into each deeper group
   via group_key[: len(source_group_key)] == source_group_key. Group () therefore holds
   all prepared execs of that outer-depth item (both o=0 and o=1), and all of them are
   copied into every per-outer group.
3. Consequence: the collector for outer iteration 0 receives outer iteration 1's item value.

Reproduced (graph: outer_iter.item -> outer_add(+1000) -> collect.item alongside nested
inner_item -> collect.item, then collect -> consumer):

• Actual: [[0, 1, 1000, 1001], [10, 11, 1000, 1001]]
• Correct: [[0, 1, 1000], [10, 11, 1001]]
• 1001 (from outer iteration 1) leaks into iteration 0's collection and vice-versa.
  Test it:

from unittest.mock import Mock
from invokeai.app.invocations.baseinvocation import InvocationContext
from invokeai.app.invocations.collections import RangeInvocation
from invokeai.app.invocations.math import AddInvocation
from invokeai.app.services.shared.graph import (
    CollectInvocation,
    Graph,
    GraphExecutionState,
    IterateInvocation,
)

from tests.test_graph_execution_state import (
    IntegerCollectionFromItemTestInvocation,
    IntegerCollectionPassthroughTestInvocation,
    create_edge,
)


def _run(g):
    while True:
        n = g.next()
        if n is None:
            break
        g.complete(n.id, n.invoke(Mock(InvocationContext)))


def test_collector_with_mixed_depth_item_inputs_leaks_across_outer_iterations():
    graph = Graph()
    graph.add_node(RangeInvocation(id="outer_range", start=0, stop=2, step=1))
    graph.add_node(IterateInvocation(id="outer_iter"))
    # Item produced DIRECTLY under the outer iterator -> path (o,) collapses to () (group key ())
    graph.add_node(AddInvocation(id="outer_add", b=1000))
    # Items produced under a NESTED inner iterator -> path (o, i) collapses to (o,)
    graph.add_node(IntegerCollectionFromItemTestInvocation(id="inner_collection"))
    graph.add_node(IterateInvocation(id="inner_iter"))
    graph.add_node(AddInvocation(id="inner_item", b=0))
    graph.add_node(CollectInvocation(id="collect"))
    graph.add_node(IntegerCollectionPassthroughTestInvocation(id="consumer"))

    graph.add_edge(create_edge("outer_range", "collection", "outer_iter", "collection"))
    graph.add_edge(create_edge("outer_iter", "item", "outer_add", "a"))
    graph.add_edge(create_edge("outer_add", "value", "collect", "item"))
    graph.add_edge(create_edge("outer_iter", "item", "inner_collection", "value"))
    graph.add_edge(create_edge("inner_collection", "collection", "inner_iter", "collection"))
    graph.add_edge(create_edge("inner_iter", "item", "inner_item", "a"))
    graph.add_edge(create_edge("inner_item", "value", "collect", "item"))
    graph.add_edge(create_edge("collect", "collection", "consumer", "collection"))

    g = GraphExecutionState(graph=graph)
    _run(g)

    consumer_prepared = g.source_prepared_mapping["consumer"]
    results = sorted(sorted(g.results[nid].collection) for nid in consumer_prepared)

    # Correct per-outer-iteration scoping:
    #   outer iter 0: inner items [0, 1]  + outer_add(0 + 1000) = 1000  -> [0, 1, 1000]
    #   outer iter 1: inner items [10, 11] + outer_add(1 + 1000) = 1001 -> [10, 11, 1001]
    assert results == [[0, 1, 1000], [10, 11, 1001]]
    # Actual behavior on this PR:
    #   [[0, 1, 1000, 1001], [10, 11, 1000, 1001]]
    #   -> outer_add's value 1001 (outer iteration 1) leaks into iteration 0's collection
    #      and 1000 (iteration 0) leaks into iteration 1's collection.

Framing: not a clean regression - on base all items land in a single fully-mixed
collector, which is also wrong. But the new grouping logic still mis-scopes this shape, and
the PR's stated goal ("preventing collections from unrelated outer iterations from being
mixed") is not met for item sources whose collapsed group key is a prefix of another group.
The collapse rule cannot distinguish "an outer-scope collection to be reused" from "an item
that should have stayed scoped to its outer iteration."

To expose this issue, add a test that feeds a collector both an item produced directly under
the outer iterator and items produced under a nested inner iterator, then asserts each
per-outer collection contains only its own outer iteration's outer-depth item (e.g.
[[0, 1, 1000], [10, 11, 1001]], not [[0, 1, 1000, 1001], [10, 11, 1000, 1001]]).

2. Low - invokeai/app/services/shared/graph.py:491-499

_get_parent_iteration_mappings_without_iterators, the matching_prepared_node is None
fallback to the () path: the reuse-a-constant-input branch is entirely untested.

• This branch is new and broadly reachable: it runs for every non-collector, iterator-free
  node, and its fallback path is what lets a downstream consumer of a nested collector pair a
  per-outer-iteration input (paths (0,), (1,)) with a constant input (path ()).
• I verified it currently produces correct output (consumer downstream of the nested
  collector plus a constant IntegerInvocation yields [[0, 1, 99], [10, 11, 99]], two exec
  nodes, is_complete=True), so this is a coverage gap rather than a proven defect - but the
  fix's correctness now depends on an untested fallback. All three added end-to-end tests use
  single-input consumers, so this two-parent path is never exercised.

To expose regressions here, add a test that places a node with two inputs (one from a nested
collector, one from a constant/top-level source) downstream of the nested-collector shape and
asserts the constant is reused across each per-outer-iteration exec node.

3. Low - invokeai/app/services/shared/graph.py:676-682

Collector branch of prepare: no coverage for a collector whose inputs yield no groups
(empty / zero-iteration collection).

• Behavior-change surface: base called create_execution_node(next_node_id, ...)
  unconditionally, so the collector branch always attempted one exec node; the new loop only
  creates exec nodes when _get_collect_iteration_mapping_groups returns groups. When a
  collector is fed only by a zero-iteration iterator, no collector exec node is created,
  next() returns None, and is_complete() stays False (stalled session) - reproduced.
• I confirmed this is pre-existing, not introduced by this PR: emulating the old
  unconditional create in the current tree still stalls, because a zero-iteration iterator
  ancestor is never added to executed, so prepare never selects the collector at all
  (guard at lines 661-665). So this is residual risk, not a regression. Listed here only
  because the PR rewrites this exact branch and adds no coverage for the empty-group path.

To characterize this behavior, add a test that runs
empty IntegerCollection -> Iterate -> Add -> Collect -> consumer and asserts the intended
outcome (either the collector emits [] and the graph completes, or the graph is explicitly
reported complete) rather than silently stalling with is_complete() == False.

[JPPhoto]

@Pfannkuchensack Responses to your points:

1. Mixed-depth item leakage: addressed

   Commit b44f5161a6 that crossed paths with your comment includes the requested failing regression test and fix. It prevents outer-iteration values from leaking into sibling collector groups.

2. Global parent reuse: addressed locally

   This was a coverage gap, not a defect. I added the requested test proving a top-level () input is reused for every nested collector iteration. It passed without further implementation changes.

3. Empty iterator/collector stall: separate pre-existing issue

   This should be its own issue with corresponding PR. Can you file an issue and tag me, please?