[merceod]

Five small graph/worker/conductor fixes that a stream-consuming node inside a Loop needs. Split out of the Nemotron VoiceChat work (#215 follow-up) so PersonaPlex can use them too.

1. Loop re-arms pure-streaming inner nodes every iteration (mstar/graph/base.py). A node whose inputs are all streams has no loop-back edge, so nothing re-injected into it after the first chunk and the loop stalled after one item.
2. The worker finds stream consumers nested in Loop / Sequential / Parallel (mstar/worker/worker.py). The consumer lookup only matched a walk that was itself a single node, so a decode loop consuming a streamed edge got no stream buffer and a KeyError when routing.
3. Stream-terminated loops stop on the final chunk (worker.py). A stream-consuming loop has no internal stop signal, so when it consumes the terminal chunk its innermost loop is stopped. Not triggered under continue_after_done, so Qwen3-Omni's talker is unaffected.
4. The conductor ends a stream-terminated partition on the worker's final-chunk signal (mstar/conductor/conductor.py). Before, producer_done for the next partition only went out when the model returned request_done, so models had to guess the end from the connection counters, which race (VoiceChat sessions lost their last audio frame). Now the worker's signal also sets request_done.
5. A request cancelled after its batch was scheduled is left out of the batch instead of failing the whole batch (Worker._build_executing_batch).

Tests: test_streaming_loop_rearm.py, test_streaming_consumer_nodes.py, test_stream_partition_done.py, test_worker_batch_skips_dropped_requests.py. ruff check . clean, test/modular green. Verified live on the VoiceChat branch (single GPU, text identical to the offline oracle, real speech, 64 concurrent sessions).

[NSagan271]

Overall looks good, but this will have to be rebased and the logic implemented with the rust graph (lmk if you want me to look into that @merceod)

[NSagan271]

Updated PR, merging in the latest version of main. The original bugs fixed by this PR changed as follows:

1. "Loop re-arms pure-streaming inner nodes every iteration": this was fixed during the rust transition already and dropped from this PR
2. Building the streaming node cache so that streaming edges can be ingested in a loop: taken unchanged
3. "The conductor ends a stream-terminated partition on the worker's final-chunk signal": taken unchanged
4. "Stream-terminated loops stop on the final chunk": ported over; now a node -> innermost loop cache is built at initialization time, which the worker uses to terminate loops on the final chunk. The original implementation used NestedLoopIdxs, which is now inside the runtime. This new version is runtime-agnostic.
5. "A request cancelled after its batch was scheduled is left out of the batch instead of failing the whole batch": at least on the latest version of main, it seems like a request cannot be cancelled after the batch is scheduled, so a KeyError can't happen on the request info. A KeyError check in retrieving the batch tensors is kept.

Also fixed a few more issues found:

1. When speculation is rolled back or a rid is dropped, the streaming chunks are returned to the corresponding stream buffers without being un-ingested from the graph node. Now the flag-setting call at submit is folded into a new commit_speculation, which in the same crossing marks the batch speculatively scheduled and un-ingests the staged streaming chunks of rolled-back or dropped rids. set_speculatively_scheduled remains for clearing the flag.
2. The new token counting happened after dereferencing tensors, so it could raise a KeyError if a tensor is freed. Now, the order of the two calls is swapped.
3. The _mark_stream_ingested docstring had incorrect rationale; updated it

[NSagan271]

Merged main in, probably worth a second set of eyes before merging