Conversation
Ports the batched surface from the draft branch, adapted to the int rid handles and uuid-only per-tensor calls that landed since. The base-class forms are loops, so every transport gets the surface for free; what they already buy is hoisting the per-call CUDA sync out of the loop. ArenaShmCommunicationManager overrides register_for_send_batch because the default would leave B host-blocking _d2h_stream.synchronize() stalls per forward pass -- the override stages the whole batch in one stream context and syncs once. register_for_send's body moves to _stage_one so both paths share it. _register_outputs now uses the batched form. The store path is left on the per-rid call: it already runs under skip_cuda_sync=True (the worker syncs on the batch completion event first), so batching it would reorder work for no gain. Also drops the unused rid parameter from set_output_ref_counts, which the rid refactor missed -- its body is entirely uuid-keyed. Co-Authored-By: Claude <noreply@anthropic.com>
…id sweep Moves set_walk, get_worker_graph_id_for_node and get_dynamic_loop_iters onto PythonGraphRuntime and deletes the WorkerGraphsManager versions. To let the port proceed method by method, the runtime is now constructed with real state and OWNS the per-request queue lifecycle; the manager is handed the same queues dict rather than building its own, so there is one copy of the state and no chance of divergence. WorkerGraphsManager add_request/remove_request no longer touch the queues. set_walk also re-derives graph_walk_worker_graph_ids, which the ported version was missing -- the walk selects which worker graphs are live, so a stale list routes a pass's inputs into the previous walk's graphs. The two get_dynamic_loop_iters call sites were per-rid loops and are now one batched call each. --- keyword drift --- The earlier rid refactor renamed identifiers with a regex that also hit keyword arguments, silently breaking nine constructors: InputSignals, ResultTensors, WorkerGraphsDone, StopLoops, RemoveRequest (x2), GraphRuntime.add_request, StreamBuffer All are on integration paths the unit suite never builds, which is why 1030 green tests said nothing. StreamBuffer is worker-internal so its field becomes rid: int; the rest are wire types and get request_id back. test_wire_keyword_arity walks the AST and checks every in-tree dataclass construction against the real fields, so the next sweep cannot do this quietly. It found the StreamBuffer case on its first run. It also found mstar/model/cosmos3/tests/* passing sampling_config= to CurrentForwardPassInfo, dropped from that type in #228. Those files are outside the suite's paths and are excluded with a pointer rather than guessed at. Co-Authored-By: Claude <noreply@anthropic.com>
Every other worker test stubs _add_new_request out, so the whole admit path was untested -- which is why eight constructor-keyword breakages from the rid sweep sat behind a green suite. Covers what the port just changed: the runtime mints the handle and owns the per-request queue lifecycle, the manager shares that same queues dict and owns per_request_info, the handle is stamped onto fwd_info, and both sides tear down on REMOVE. The teardown assertion matters because handles are recycled -- the test admits a second request and asserts it gets the same integer back, so anything left behind would attach to it. Verified against the real regression: reintroducing add_request(rid=...) fails two of the three tests. Also scopes test_wire_keyword_arity to mstar-owned dataclasses. HF configs (PretrainedConfig subclasses) take **kwargs, so dataclasses.fields() does not describe what their __init__ accepts and every extra kwarg read as an error. Co-Authored-By: Claude <noreply@anthropic.com>
check_dyn_loop, get_dyn_loop_workers and the sharding-config accessor move to PythonGraphRuntime and the WorkerGraphsManager versions are deleted. These are pure reads over dyn_loop_to_workers / sharding_config, which add_request already builds on the runtime -- the manager's PerRequestInfo was carrying a second copy of the same thing. Also stores the communicator param (added to the signature but not kept), which stop_loops_batched and send_outputs will need. Fixes a shadowing bug in the loop-stop block while converting it: the inner `for worker, loop_names in stop_loop_workers.items()` rebound the set the enclosing iteration was still working from. Harmless today because it is the last use, but it is the same shape as the bug ruff caught in the add_request port. Co-Authored-By: Claude <noreply@anthropic.com>
Completes the bookkeeper's tensor-info storage: adds the read side (get_info / get_info_batch) and threads the descriptor through every put_tensor call site. This is what unblocks the flat ingest/route contract. Those carry uuids, but a disaggregated loop re-emits its ingested external inputs as outputs, which can be routed to a peer -- and there the descriptor IS the payload the peer reads from. A uuid-only stub would produce a structurally valid but unreadable message, failing as a garbage remote read rather than an exception. The stored descriptor is the same object the edge carries, so the in-place shm_segment/shm_offset that register_for_send stamps on is visible through both. Two sites needed reordering to bind the descriptor before the put: store_and_return_tensor_info built it after, and the slice path re-points info.uuid at a freshly minted uuid -- moving the put after the re-point makes the stored descriptor name the slice rather than the producer's tensor by construction, instead of relying on it being a shared reference. The last test pins the whole premise end to end: an edge holding nothing but store.get_info(uuid) is read correctly by a peer over SHM. Co-Authored-By: Claude <noreply@anthropic.com>
Uses the last_node_run parameter you added. loop_stop_times moves off CurrentForwardPassInfo onto GraphRuntimeRequestInfo per the note: it was worker-only state riding on an object that gets forwarded to peers. clear_loop_stop_info went with it -- it had no callers. _pending_loop_stops moves to the runtime too, since step (2) of the contract puts it there. Its three worker readers become accessors. The WorkerGraphsManager.stop_loops and the worker's inline stop block are deleted; the runtime now does the check_dyn_loop filtering, the graph stop, the stop-time snapshot and the peer fan-out in one call. --- added ABC surface, please review the shape --- Moving that state required five methods the contract did not name: apply_peer_loop_stops, get_loop_stop_times, has_pending_loop_stop, pending_loop_stop_rids, clear_pending_loop_stops. PendingLoopStop moved from worker.py to runtime/base.py for the same reason. apply_peer_loop_stops is the one worth a look: the receive side of STOP_LOOPS needs to be on the contract for a Rust backend to handle the message at all, and it is deliberately NOT stop_loops_batched -- it takes no last_node_run (the originating rank took the snapshot) and does not fan out (that rank already told everyone), so routing it through the local path would loop. Tests move with the behaviour. The peer-merge tests initially passed while constructing a malformed NestedLoopIndices (loop_indices is a dict, not a list) because wg_fwd_pass_idx short-circuits label_context_gt before that field is read; fixed, and added a case that holds the forward pass constant so the enclosing-loop-index comparison is actually exercised. Co-Authored-By: Claude <noreply@anthropic.com>
get_ready_nodes and has_ready_excluding move to PythonGraphRuntime; both
get_next_batch's scan and the speculation peek now go through them.
The split the contract asks for: the runtime answers "whose graph inputs
are satisfied", the scheduler applies engine readiness. Engine readiness
has to stay outside because _check_ready can FAIL a request (an
AdmitRuntimeError parks it in admit_errors), which is a scheduling
decision, not a graph one.
Two asymmetries between the old scans are preserved deliberately rather
than unified:
- get_next_batch filters to parallel_leader_nodes (only rank 0 initiates);
the peek does not. The filter stays in the scheduler so the runtime's
scan means the same thing for both callers.
- the peek counts a rid in the head TP-follow batch as ready even when
failed, because get_next_batch will schedule that one regardless. That
stays in the exclude set the scheduler builds.
has_ready_excluding is now used as an early-out: graph readiness is
necessary but not sufficient, so a False is final and skips the engine
pass entirely, while a True falls through to the engine-filtered
confirmation. That keeps the cheap first-match scan meaningful without
changing what the method reports.
Also drops get_loop_stop_times from the ABC -- its only caller is the
runtime's own fan-out, so it becomes _loop_stop_times. The other four
added methods are all called from worker.py and stay.
Test fakes gain the graph-level scan over the same queues the manager
holds, mirroring the real sharing.
Co-Authored-By: Claude <noreply@anthropic.com>
Finishes the scheduling group. Both scheduler pop sites (pop_ready_rids and _assemble_batch) and all four worker push-back sites go through the runtime; WorkerGraphsManager.push_back_node is deleted. pop_rids returns PopRidsOutput, so it also hands back the batch's ready inputs as EdgeSpecs. Nothing consumes those yet -- _build_executing_batch still walks node.ready_signals -- so there is a TODO at the call site rather than a second path. That switch is what makes the descriptor store earn its keep, and it belongs with complete_and_route_batch. Engine readiness stays ahead of the pop in the scheduler: the contract says pop_rids assumes it, and it has to run over the whole set first because both checks are all-or-nothing. get_nodes is explicitly transitional and says so. The worker still carries GraphNodes through its batch for postprocess; those uses move into complete_and_route_batch and this goes with them, since a Rust backend cannot hand out live Python graph objects. --- ParallelList footgun --- `batch_rids, wg_ids = popped.wg_ids` looks like NamedTuple field unpacking but goes through the overridden __iter__, so it binds (key, value) PAIRS. At exactly two entries it succeeds silently and binds the wrong thing; here it happened to raise. Documented on __iter__ and pinned in test_containers.py, including the two-entry case. Co-Authored-By: Claude <noreply@anthropic.com>
process_node_outputs, mark_node_complete and get_nested_loop_idxs_for_node move to PythonGraphRuntime and the WorkerGraphsManager copies are deleted. Everything process_node_outputs needed was already on the runtime (sharding_config, node_to_workers, worker_graph_ids, the walk->wg index, the queues), so it is a faithful move with field accesses retargeted. The worker's postprocess prologue now stores tensors and hands over uuids: store_and_populate_graph_edges splits back into store_and_return_tensor_info plus a safety-hold increment, and the runtime populates its own edges from tensor_store.get_info. That is what the descriptor store was for. RouteOutput drives the staging: the runtime says which tensors remote consumers will read (deduped by uuid, skipping already-registered), and _register_outputs is now a translation from indices back to descriptors instead of re-deriving the set from the routing. take_completion is transitional and labelled: send_outputs will consume the parked routing internally, and it goes away then. --- test that proved nothing --- test_route_batch_decodes_the_flat_rid_major_layout initially used ONE output signal, so num_tensors[i * n_signals + s] and a transposed num_tensors[s * n_rids + i] index identically -- it passed against a deliberately transposed implementation. Rewritten with two signals and asymmetric per-rid counts, and re-verified: the transposition now fails it. Same failure mode as the NestedLoopIndices case two commits ago, so it is worth being systematic about checking that these tests can fail. Co-Authored-By: Claude <noreply@anthropic.com>
The runtime now owns the sends and the per-request buffers: persist
signals, new-token counts, output chunk names and output loop indices move
off PerRequestInfo, and the six buffer_/flush_ methods plus the loop-index
pair are deleted from WorkerGraphsManager.
What stays worker-side is what touches real tensors or cannot be derived at
send time, handed over via SendInput:
- new-token counts need numel(), so the worker counts and passes them
- local streaming feeds a StreamBuffer, which holds tensors; RouteOutput's
local_streaming_tensor_idxs is how the caller finds those
- stream_tokens_consumed, partition_done_rids, profiling (msgpacked, so
the runtime need not own RxInfo/TxInfo)
Adds nested_loop_indices to SendInput. The snapshot has to predate the
completion -- mark_node_complete and stop_loops both advance loop state --
so the runtime cannot re-derive it at send time. Same category as the other
"cannot derive" fields already there.
take_completion becomes peek_completion: send_outputs is what pops the
entry, and the worker still needs to see the routing first. As written the
two both popped, so the second raised KeyError -- on the live path only,
which no test reached. test_send_outputs_consumes_the_completion now covers
the handoff.
--- second ParallelList footgun ---
`dict(pl)` raises "'list' object is not callable": dict probes for a
.keys() METHOD to decide it was handed a mapping, and ParallelList.keys is
a list attribute, so the probe calls it. Needs dict(iter(pl)). Documented
on the class alongside the unpacking trap and pinned in test_containers.
Co-Authored-By: Claude <noreply@anthropic.com>
All four ingest sites (new request, peer INPUT_SIGNALS, stream-buffer poll, ready-tensor poll) go through the runtime. WorkerGraphsManager's process_new_inputs / process_new_streaming_inputs are deleted, and so is the WorkerGraphQueues streaming variant -- the runtime applies the ready_for_streaming gate inline, per signal, so ingesting one signal can be what makes the next eligible. Only uuids cross the boundary; the runtime rebuilds descriptors from the tensor store. Checked that all four provenances have them: signal-only edges carry none, peer edges get theirs from start_read_tensors, and the streaming synthetic edge mints fresh ones through store_and_return_tensor_info. The return type goes from leftover GraphEdges to indices, which is what the streaming path actually needs -- it hands the original edge back to its StreamBuffer, so an index is enough and the edge never has to survive the round trip. Both test stubs gain a real TensorStore, since resolving descriptors is now part of what a tensor manager has to provide. Co-Authored-By: Claude <noreply@anthropic.com>
set_node_metadata had no caller anywhere in the tree, so the runtime's parallel_nodes / parallel_leader_nodes / tp_async_nodes were all empty. Nothing depended on them yet, so it was latent rather than broken -- but speculate_node is the first consumer and would have silently mis-filtered. The worker resolves two cases the contract's signature cannot express before passing them down: tp_async_nodes=None means "every node", and the feature being off means "none". The runtime just takes the effective set, where empty means none. speculate_node is implemented and tested but NOT yet wired into _try_speculate_next, because the two speculation methods are coupled: the worker's per-rid filtering reads is_new_loop_iter and loop_name off SpeculativeNodeInfo, and SpeculationOutput deliberately drops both since the contract moves that filtering into prep_spec_rids. Converting the call site without prep_spec_rids would lose it. Calling ingest_for_speculation twice to recover the fields is not an option either -- it mutates speculative slot state. Co-Authored-By: Claude <noreply@anthropic.com>
prep_spec_rids does the per-rid loop-completion filtering as designed (pending stop, final iteration), the streaming ingest with slot tracking, the readiness check and the rollback. SpeculationPrepOutput gains wg_ids. The worker keeps what holds tensors: polling the StreamBuffers, returning unconsumed chunks to them, and turning the returned EdgeSpecs into tensors. --- two contract notes --- SpeculationOutput gains is_new_loop_iter and loop_name. NOT for filtering -- prep_spec_rids still owns that -- but the speculative batch carries them onto PendingBatch, and the postprocess needs them to match its pending loop stops. The selection step already computes both, so reporting them avoids a second ingest_for_speculation, which would mutate speculative slot state. The TP-follow path still preps through the worker-side helper. A follower is all-or-nothing: rank 0 committed to that exact composition and sits on the collective until every follower joins, so one rid failing has to roll the whole batch back. prep_spec_rids is per-rid best-effort and the input has no way to ask for the other mode. Flagged rather than guessed at. --- a test asserting something structurally impossible --- The two loop-filter tests first drove prefill -> ar_decode, which ENTERS the loop rather than looping back, so is_new_loop_iter is False and the filter cannot fire. They failed, which is how the fixture error surfaced; rewritten to speculate ar_decode -> ar_decode, with a sanity assertion that the loop-back is speculatable before each stop is applied, plus a case pinning that the filter correctly does NOT apply on the way in. Co-Authored-By: Claude <noreply@anthropic.com>
Completes your WIP. ScheduledBatch now carries only rids, worker graph ids
and the ready inputs the pop already reported; no GraphNode survives in
flight, so get_nodes and _get_wgio_for_rid are gone.
Per your four decisions:
(4) pop_rids' input_edges thread through ScheduledBatch, so the batch build
and both fresh-rid merges read them from there. _get_input_tensors is
deleted; _tensors_for is the one place uuids become tensors.
(2) get_consumed_edges is a standalone runtime query -- structural, so it
takes no rid.
(3) prep_follow_spec_rids is separate, sharing _prep_one_spec_rid and
_streaming_edges_by_rid. The undo is factored out as _undo_spec_ingest
so it serves both the per-rid not-ready case and the follower's
all-or-nothing unwind.
(1) _partition_done is internal: the worker calls mark_stream_partition_done
and send_outputs derives the rest, so SendInput loses
partition_done_rids.
Also added get_spec_target: a follower cannot use speculate_node, because
that filter requires the node be in parallel_leader_nodes and a follower by
definition is not. The leader already chose; this reports the target's loop
context.
The follower now pops fresh rids BEFORE prepping continuing ones. The old
order needed _rollback_all when the pop failed; reversed, the only undo a
failure needs is push_back_node, which already exists.
Four bugs in the WIP:
- pop_ready_rids returned a 1-tuple (trailing comma), so the TP-follow
path would have put a tuple in request_to_worker_graph
- `node` was undefined in the leader's fresh-rid merge
- _try_follow_speculate still called _assemble_speculation with the old
arity; ruff missed it because sample_node was bound in that scope
- request_to_worker_graph kept `= None` after node_objects (the required
field) was removed, making an empty ScheduledBatch constructible and
__len__ raise; now default_factory=dict
Co-Authored-By: Claude <noreply@anthropic.com>
It stopped managing worker graphs several commits ago; this makes the name
and the contents match. What is left is the request state the runtime
deliberately does not take: CurrentForwardPassInfo, which it treats as an
opaque wire object, and the StreamBuffers, which hold tensors.
Deleted as dead or duplicated by PythonGraphRuntime:
PerRequestInfo node_to_workers, dyn_loop_to_workers, worker_graph_ids,
sharding_config
PerPartitionInfo graph_walk_worker_graph_ids, stream_partition_done
manager queues, base_sharding_config, the three
all_worker_graph_ids_to_* maps, walk_node_to_worker_graph_id
and the __post_init__ that built it, get_graph_walk,
get_publish_info, get_fwd_number, has_partition
add_request drops from five parameters to two: the queue lifecycle and the
routing tables both belong to the runtime, so registering a partition is
all that is left.
The four sharding_config readers now go through the runtime, which is the
last thing the manager's copy was for. get_sharding_config returns None for
an unknown rid -- the drain and TP-fanout callers can legitimately race a
removal, and they already guarded for it.
The shared-queues wiring is gone with it: the manager was handed the
runtime's dict so the port could proceed method by method, and nothing on
that side reads queues any more. The admit test asserts the finished split
instead.
test_worker_graphs_manager.py is now test_graph_runtime.py, which is what
it has been testing for a while.
Co-Authored-By: Claude <noreply@anthropic.com>
Adds graph/request.rs following the struct-per-concern shape: RequestPartitionInfo (walk, the worker graphs it selects, stream-done), RequestInfo (partitions, node/loop -> workers, loop stop times), and WorkerGraphMeta for the static worker-graph facts. Partition info is keyed by partition on the request, NOT stored in the worker graph queues -- see the module doc. A partition spans several worker graphs, so putting it there duplicates the walk and makes set_walk a scatter-write. The worker graph -> partition direction IS static, so that lives on WorkerGraphMeta and costs nothing per request. set_node_metadata and add_request are filled in, plus remove_request, get_rid_handle/string, get_walk, set_walk and mark_stream_partition_done. InternedRids no longer carries its own copy of the walk state -- the runtime owns states[wg][handle]. intern() now reports whether the handle is new so the caller knows to grow its per-handle vectors, and release() is idempotent: a double free would push one handle onto the free list twice and hand it to two requests at once. add_request needs the workers behind REMOTE worker graphs to build node_to_workers, which the constructor never saw -- it only takes this worker's graphs. set_remote_worker_graphs takes the conductor's global maps; the wg_ids already present are skipped. Tests cover the recycling hazard (a leaked handle-keyed entry reads as one request inheriting another's state, not as staleness), set_walk's transition detection and cache refresh, and label_context_gt including the same-forward-pass case where it must fall through to the enclosing loop indices rather than short-circuit. Co-Authored-By: Claude <noreply@anthropic.com>
ShardingConfig splits into ShardingTemplate (the deployment-wide config the conductor hands over) and ShardMap (one request's resolved binding). instantiate() is clone_empty() followed by setup(node_to_workers) in one step, and the result hangs off RequestInfo. Per request because a data-parallel replica puts the same node on different workers, so node -> workers is not deployment-wide. The prototype's single global ShardMap would have routed every replica to the first one's workers. Keys become (node, Option<walk>) rather than node alone. The None entry is Python's streaming lookup: a streaming consumer's walk is not known when the edge is routed, so setup adds it for any group that spans all walks, and for a singleton whose node has exactly one worker-set combination. Both halves of setup are ported: configured groups claim their nodes first, then whatever is left becomes a singleton per (node, worker set) at tp_size 1 -- so several walks sharing a worker set share one group rather than getting one each. The two asserts become a ShardError that add_request surfaces as a ValueError: a worker count disagreeing with tp_size, and two groups both claiming a node with graph_walks=None. Both are deployment config errors, worth failing loudly rather than routing around. Tests cover the streaming-None entry appearing and NOT appearing (two walk combinations make it ambiguous, which is why Python omits it), the singleton fallback sharing a group across walks, and two requests binding the same node to different workers -- the case the whole change exists for. Co-Authored-By: Claude <noreply@anthropic.com>
…uction The constructor was synthesising a single sharding group from `workers`, which was a prototype stand-in: it had no shard_dim, no per-group nodes or graph walks, and no tp/sp_enabled_nodes. ShardingArg now carries all of it and `workers` is gone -- group membership comes from the config, and the conductor already sets each group's tp_rank per worker. shard_dim's Python type is `dict[str, int | None]`, but a None value reads exactly like an absent key (both mean replicated), so the Nones are dropped on the way in rather than carrying an Option nothing distinguishes. tp_enabled_nodes / sp_enabled_nodes gate config validation only -- they add no shard_dim entries and do not affect routing -- but the type now carries them, so it is the whole ShardingConfig rather than the part routing reads. Remote worker graphs move into the constructor and set_remote_worker_graphs is deleted; add_request cannot build node_to_workers correctly without them, so a separate call was a sequencing trap. The new test covers shard_dim reaching fanout, because losing it does not fail: it silently takes the replicated branch and sends whole tensors where slices were meant. Co-Authored-By: Claude <noreply@anthropic.com>
Fills in tensors.rs: descriptors plus the full reference surface
(put/update/get info, ref counts, persist, mem_registered, can_gc,
collectable) and the batch forms.
TensorPointerInfo's strings are interned. dtype, source_entity,
source_session_id and shm_segment come from a handful of values but repeat
on every tensor, so they are ids inside and strings only at the boundary.
dtype is the wire codec's short name ("float16"), so Python converts with
the _dtype_name / _dtype_from_name it already has rather than a second
mapping.
Semantics kept from Python: mutating an untracked uuid is a no-op (a late
TENSOR_RECEIVED for an already-collected tensor is benign), put_tensor
RESETS reference state (a live uuid means a new tensor; update_info is the
edit), increment_ref rejects a negative n as Python asserts, and
dereference accepts one because set_output_ref_counts corrects downward
that way.
get_info returns a #[pyclass] rather than a tuple: pyo3 only converts
tuples up to 12 elements and this has 15, and named fields make the
reconstruction on the Python side obvious.
--- build config ---
extension-module becomes a default feature instead of being unconditional.
A Python extension must NOT link libpython (symbols resolve at import),
but a test binary is an executable and has to, so `cargo test` could not
link once #[pyclass] pulled in more of the C API. Tests now run with
`cargo test --no-default-features`; the cdylib is unchanged.
--- python ---
PythonGraphRuntime.queues existed to share the dict with the manager, which
no longer holds one. No production caller remained; the tests reach
_queues directly, as they already do for the other internals.
Co-Authored-By: Claude <noreply@anthropic.com>
The batch signatures take two parallel lists instead of a ParallelList, which is what actually crosses the boundary -- pyo3 converts a list of ints directly, where a NamedTuple needs unpacking per call. RustTensorBookkeeping is a drop-in for PythonTensorBookkeeping, so TensorStore does not know which it holds. Descriptors convert at the seam: dtype uses the wire codec's short name, so there is one torch.dtype mapping rather than two. One behavioural difference is documented on the class: Rust COPIES a descriptor in, where the Python backend keeps the caller's object. Anything relying on seeing a later in-place edit (register_for_send stamping shm_segment) has to re-update_info. Nothing does today -- the Python path reads the store's copy -- but it would fail as a wrong address rather than an exception, so it is worth stating. test_rust_tensor_bookkeeping runs every case against BOTH backends, since a divergence is silent: a refcount that drops early frees a tensor a peer is still reading, one that never drops leaks until the request ends. Verified the parity actually bites -- dropping put_tensor's reset fails test_put_tensor_resets_reference_state on the rust backend only. --- two bugs in the new set_speculatively_scheduled --- It looked up the node with interner.get(), which is the STRING id, and passed it where a node index within the worker graph was wanted. Both are u32, so only the Option in the signature made the mix-up visible; had it been unwrapped it would have flagged whichever node happened to sit at that index. Now uses nid(wg, name). It also indexed states[] with the wire worker-graph id, which is the conductor's dense index over the deployment, not this worker's local position. Added wg_index() to map between them. Co-Authored-By: Claude <noreply@anthropic.com>
GraphRuntime takes a share of the TensorBookkeeping at construction, as asked. TensorBookkeeping splits into the state (Bookkeeping) and the #[pyclass] handle, both holding Arc<Mutex<_>>, so GraphRuntime gets the SAME bookkeeper Python handed TensorStore. A copy would diverge the moment either side moved a refcount, and nothing would raise. Rust gains the structural lookups (get_worker_graph_id_for_node, is_async_schedulable, get_output_signals, get_consumed_edges), the pending loop stops and push_back_node. node_owner was being built in the constructor and discarded; it is kept now. GraphRuntime is also registered in lib.rs -- it was not, so it was unreachable from Python. graph/runtime/rust.py is the Python side: the compile seam (worker_graph_args / sharding_args) plus the ABC over mstar_rust. The 15 unported methods are bound in the CLASS BODY rather than setattr'd, because ABCMeta freezes __abstractmethods__ at class creation and a later assignment leaves the class abstract; each raises naming itself. MSTAR_RUST_GRAPH=0|1 selects the runtime. No AUTO: the Rust one cannot run a forward pass yet, so falling back silently would hide that and picking it up silently would break a worker that merely has the extension installed. It requires MSTAR_RUST_ZMQ != 0 and a RustZMQCommunicator, since the runtime sends from Rust over a shared transport, and a Rust bookkeeper, since it holds a share of it -- each checked with its own message. Tests cover the compile seam specifically: a node compiling into the wrong worker graph routes its outputs to the wrong peer, silently. Co-Authored-By: Claude <noreply@anthropic.com>
…rship Your three corrections, plus one the second turned up. _managing_registry is only populated by a WorkerGraphIO, which the worker builds per request -- so compiling straight off worker_graph.section saw None every time and every loop crossed as top-level, accounting a nested loop's completion to the wrong registry. The seam now builds one for that side effect, over a deepcopy: WorkerGraphIO writes the registry into the sections it walks, and the pristine section is shared by every request. accumulated_outputs is a real Loop field, so it is read directly rather than through a getattr default that would have silently sent nothing. leader_nodes is gone from WorkerGraphArg and compile_one. Leadership is per worker, not per worker graph, and already arrives via set_node_metadata; NodeSpec.is_leader was written and never read, so it went too. --- found while fixing the second --- streaming_inputs was read off GraphNode, but the field there is _streaming_inputs; the public name belongs to ReadySignals. The getattr default meant every streaming node crossed with an empty set, losing its ready-for-streaming seed. It is populated by _register_streaming during worker-graph construction, so it is set by the time we compile. The getattr defaults were the common cause in all three: each turned a wrong attribute name into a plausible empty value instead of an error. Direct access now, so the next one raises. Tests pin the loop nesting, that compiling leaves the shared section untouched, and that streaming inputs cross. Verified the nesting test fails without the WorkerGraphIO. Co-Authored-By: Claude <noreply@anthropic.com>
Offers each signal to the live worker graphs until one claims it -- a claim loop, not a lookup, because a node can refuse a signal it owns when the name is not one of its inputs or both ready slots are full. Descriptors resolve once per signal through the shared bookkeeper, not per worker graph. A uuid with no descriptor keeps its identity and zeroes the shape: the uuid is what routing keys on, and Python tolerates the same case (get_info returning None inside a tensor_info list). The streaming gate is re-checked per signal, since ingesting one can be what makes the next node eligible -- the contract calls this out and a hoisted check would drop the second chunk of a pair. Adds RequestState::is_ready_for_streaming and Bookkeeping::tensor_ref. Tests cover the refusals specifically, since each returns an index rather than raising: unknown node, an input the node does not take, both slots full, can_buffer=False, an unknown rid, and the streaming gate. Co-Authored-By: Claude <noreply@anthropic.com>
get_dynamic_loop_iters, cleanup_consumed_inputs and reset_outputs. cleanup_consumed_inputs releases the just-executed node's inputs and dereferences them in the bookkeeper the store shares. A loop's external inputs are EXCLUDED: they are re-injected each iteration (Python's _persist_for_loop), and clearing them would strand the loop waiting for a signal nobody resends. That exclusion is structural -- from the spec's external_inputs -- rather than a per-request flag, which is why Rust can derive it without tracking the flag Python sets during ingest. reset_outputs is a no-op by construction and says so: Python clears tensor_info off per-request edge objects that carry it, where here outputs are passed into complete_and_route_batch, so there is nothing stale. The Bookkeeping operations become pub -- GraphRuntime calls them directly now, not only through the pyclass. Co-Authored-By: Claude <noreply@anthropic.com>
get_ready_nodes, has_ready_excluding and pop_rids, over one scan_ready walker whose callback returns false to stop -- which is what lets the peek bail on the first match instead of building the whole list, the reason the contract has both. Graph level only. Engine readiness stays with the caller, because _check_ready can FAIL a request, and that is a scheduling decision rather than a graph one. pop_rids with check_ready verifies every rid before popping any, so one not-ready rid leaves the set intact for a later retry rather than half consuming it. Its input_edges come straight from the slots it just took, so _build_executing_batch never has to walk the ready signals again. Co-Authored-By: Claude <noreply@anthropic.com>
…ignature test_runtime_parity drives BOTH runtimes through the same sequences and compares every answer, rather than asserting hand-written expectations twice. A divergence here is silent in production -- a request scheduled on one and not the other, a refcount off by one -- so running one script against both is the check worth having. Covers admit/ingest/schedule, the structural lookups, every ingest refusal, handle recycling, speculation (including is_new_loop_iter both ways), routing with the safety hold, the prep room cap, and input cleanup. Also fixes GraphRuntime.add_request's signature: it declared `worker_graph_to_worker: ParallelList[int, str]` where both implementations take `worker_graph_to_workers: ParallelList[int, list[str]]`. The name differs, so a keyword call written against the ABC would have failed. Co-Authored-By: Claude <noreply@anthropic.com>
A handle can outlive its request -- a message for a rid this rank already removed is a benign race the Python side has always tolerated with .get(). Five Rust entry points indexed states[wg][rid] directly and panicked ACROSS the FFI boundary instead: cleanup_consumed_inputs, push_back_node, set_speculatively_scheduled, prep_spec_rids and prep_follow_spec_rids. A PanicException is worse than an exception here: the worker's error handling does not expect it, and unwinding through the boundary is not something to rely on. All state access now goes through bounds-checked state() / state_mut(). The three remaining direct indexes are in add_request and remove_request, where the vectors were just grown for that handle, and say so. Found by fuzzing every method with an out-of-range handle; that fuzz is now a parametrised test over all 19 entry points, so a new one cannot quietly reintroduce the pattern. Co-Authored-By: Claude <noreply@anthropic.com>
PythonGraphRuntime.add_request minted a fresh handle on every call. The conductor sends one NewRequest PER PARTITION, so a three-partition model (qwen3_omni: Thinker / Talker / Code2Wav) got three handles for one request. Only the last was reachable through _rid_to_handle, so remove_request freed only that one. The other two kept their per-request queues alive -- each a deepcopy of the whole graph section -- with no way left to reach them. Two leaked graphs per request, growing without bound, plus their tensor references never dereferenced. Found by the differential test: Rust returned one handle where Python returned three. The Rust side already did this correctly. Co-Authored-By: Claude <noreply@anthropic.com>
RequestState::complete clears a top-level node's ready slot -- matching Python, which only clears for top-level entities -- but did so without reporting the uuids, so nothing dereferenced them. The worker cleans up BEFORE routing, so on that path the slot is already empty and the counts came out right. The leak only appears if completion runs first, which made it invisible to every test written in worker order. Python has no such dependence: mark_entity_complete clears THROUGH the tensor manager, so the dereference happens either way. complete now returns what it freed and complete_and_route_batch dereferences it. The test runs both orders. Co-Authored-By: Claude <noreply@anthropic.com>
complete_and_route_batch parks the routing for send_outputs to consume. An exception between the two -- _register_outputs raising, a missing tensor in the new-token count -- abandons it, in BOTH runtimes. That is not merely a leak. Handles are recycled, so the parked entry still names a rid; the next request to get that integer would have another request's outputs sent under its name. remove_request now drops the rid from any parked completion and discards one left empty, which is the same rule as every other handle-keyed map. Rust gains num_parked_completions so the invariant is observable: a number that only grows means sends are being abandoned. Co-Authored-By: Claude <noreply@anthropic.com>
The postprocess prologue took the ref=1 hold one tensor at a time. With a Rust bookkeeper each of those is a boundary crossing, and a 128-request batch with two output signals makes 256 of them: 17.6us against 4.7us for the same work batched, 3.8x. Python is unchanged, having no boundary. increment_ref_batch existed on TensorBookkeeping but not on TensorStore or the communication manager, so the call site could not reach it. Co-Authored-By: Claude <noreply@anthropic.com>
Three problems with the wiring, on top of the one you found. The factory gated on MSTAR_SHM_ARENA, which selects the tensor TRANSPORT and has nothing to do with the graph runtime. It is MSTAR_RUST_GRAPH that needs the Rust bookkeeper: the runtime holds a SHARE of that object, so the two have to be the same implementation. Nothing else should select it -- a descriptor is copied in and rebuilt out, which the Python runtime pays for and gains nothing from. `from mstar_rust import TensorBookkeeping` sat at tensor_store's module scope, and tensor_store is imported by effectively everything -- so the whole package became unimportable on any machine that had not run maturin. Now imported inside RustTensorBookkeeping.__init__, the same pattern arena.py uses. The three test files still imported the deleted rust_tensor_store module, and their importorskip probed the WRAPPER rather than the extension. Since the wrapper now imports fine either way, that would have failed rather than skipped where the extension is absent; they probe mstar_rust itself. test_tensor_store_backend pins all of it, including importing the package with the extension blocked at the meta-path -- the simulation has to raise ModuleNotFoundError, since pytest's importorskip deliberately does not skip on a bare ImportError. Co-Authored-By: Claude <noreply@anthropic.com>
Staging stamps shm_segment/shm_offset onto the TensorPointerInfo in place. That reached the store only because the Python bookkeeper hands back the caller's own object -- the Rust one copies the descriptor in, so the stamp died on a throwaway rebuild. The wire descriptor is looked up by uuid, not carried: RouteOutput gives indices, and rust.py::_infos rebuilds from the bookkeeper. So under MSTAR_RUST_GRAPH=1 every arena-staged tensor shipped with shm_segment=None, which the consumer reads as "spilled to a per-uuid file" -- and goes looking for a file that was never written, because the bytes went to the arena. Made the stamper's job rather than the caller's; register_for_send has two callers and neither should have to know. Collected across the batch so it is one crossing per pass, not one per tensor. The existing roundtrip test only checked the stamp on the caller's object, which passes either way. The new one reads it back out of the store, over both bookkeepers and both register forms; without the writeback the two Rust cases fail and the two Python ones pass. Co-Authored-By: Claude <noreply@anthropic.com>
The rid sweep renamed the tensor manager's parameter to `rid`; the preprocess worker's call sites kept `request_id=`. You caught three. `ack_unread_tensors` at data_worker.py:462 is the fourth -- the late-tensor ack for a request that is already gone, so it fires exactly when a request aborts mid-read and the producer is waiting to free its buffers. All four are a TypeError the moment a request carries a tensor, and none is reachable from the unit suite: the preprocess worker runs on its own thread against a live mesh. The guard resolves the class behind a named receiver and checks keywords against the real signature. Scoped to the four rid seams on purpose -- the first attempt unioned every def by method name, which flags every `.gather(dim=...)` on a torch tensor because some unrelated mstar method is also called `gather`. Reverting any of the four fixes fails it. TensorStore's rid is now typed `int | str`. The preprocess worker never took the int refactor and does not need to: rid is an opaque dict key here, never arithmetic, and it does not reach the bookkeeper at all. Co-Authored-By: Claude <noreply@anthropic.com>
Every worker test injects a runtime directly, so nothing checked that MSTAR_RUST_GRAPH=1 reaches the Rust one -- which is how the Rust bookkeeper shipped never being instantiated. A flag that silently does nothing reads exactly like a flag that works. Covers both selections, the refusal of anything that is not 0 or 1, all three guards, and that the runtime holds the SAME bookkeeper object rather than a rebuilt one. Co-Authored-By: Claude <noreply@anthropic.com>
…ffers Two divergences from the Python runtime. speculate_node filtered on the SOURCE node's enable_async_scheduling. The caller already filtered the source (worker.py::_can_speculate), so the real check -- the destination's -- was simply missing, and a node that opted out got speculated into and then dropped per rid. The parity fixture could not see it: every node in that graph is async-enabled, so the source-side and destination-side checks agree on every input it produces. The new fixture opts the target out; without the fix the Rust case fails and Python passes. remove_request forwarded to Rust and left four Python-side dicts keyed by the handle. Handles are recycled, so a request aborted between the route and its worker-graph completion would hand its persist signals, token counts and buffered output names to whichever request next drew that integer -- exactly the hazard base.py documents. _output_loop_indices was never popped at all. Co-Authored-By: Claude <noreply@anthropic.com>
get_sharding_config was `pass # TODO`, so it returned None and the very first NEW_REQUEST handed None to register_request -- MSTAR_RUST_GRAPH=1 could not admit a request. Derived in Python rather than exposed from Rust. Rust does build its own copy for routing, but it cannot come back out usefully: register_request wants the Python ShardingConfig and the four TP fan-out sites read group._workers, so returning it would mean reconstructing the object across the boundary -- a SECOND derivation, in a new place, free to drift. Deriving from the same input by the same rule cannot. Nothing is replicated: clone_empty()/setup() IS the logic, called once. The one piece of assembly both runtimes were doing -- building node_to_workers from the per-worker-graph map -- moved to graph/runtime/sharding.py and python.py now calls it too. Parity tests pin that the two agree, that the base config is never handed back, that removal drops it (handles are recycled), and that an unknown rid answers None rather than raising -- teardown and TP fan-out race removals. Still unserved on the Rust path, both needing a design call: get_nested_loop_idxs_for_node, and peek_completion (the worker reads GraphEdge objects off it where Rust has only indices). Co-Authored-By: Claude <noreply@anthropic.com>
Python sweeps all worker graphs the request runs in this walk and resets each done one; Rust checked only the graph owning the completed node and never reset anything -- RequestState::reset had zero call sites. Two consequences. A worker graph that became done without ingesting an edge in this call was silently dropped: the case Python calls out for Orpheus prefill, BAGEL vae_decoder and Code2Wav, where the completed node's outputs all go to EMPTY_DESTINATION / EMIT_TO_CLIENT / a streaming partition. And `is_done` latched, so root_entity_done early-returned on it, node flags and loop counters never cleared, and a request reported done exactly once. A worker graph completes many times over one request, so the second effect is every request past its first pass. The parity fixture runs each request through exactly one pass, which is why it could not see this. The new one is a single node that finishes its graph in one completion, driven twice; without the reset the second ingest goes nowhere and the Rust case fails. Co-Authored-By: Claude <noreply@anthropic.com>
Python settles the refcount over the POST-fanout edges -- to_workers is keyed by worker, so a tensor read by N workers is counted N times. Rust counted the pre-fanout list, one entry per graph edge; the per-worker expansion happens later, in take_send_plan, after the counts are already settled. So with a TP consumer the hold dropped to 1 while N reads were outstanding, and the first release freed a tensor the other N-1 were still reading. ShardMap::fanout had no caller outside its own unit tests. EMPTY_DESTINATION was counted as 1 in the same loop. Python drops it before the count -- it routes nowhere -- so that was a reference no one would ever release. Every other fixture puts one worker on each node, which makes an edge and a destination indistinguishable; the new one gives the consumer a real TP group of two. Without the fix the Rust case frees after the first release. Co-Authored-By: Claude <noreply@anthropic.com>
…_node is_first_tp_rank was hardcoded True in the Python shim, so under TP>1 every rank claimed rank 0 and the conductor counted tp_size reports per request instead of one. Rust has what it needs -- the per-request ShardMap -- so it is computed at completion and rides the plan. Computed at COMPLETION, not at send: the group lookup is per completed node. get_nested_loop_idxs_for_node was Python-only, an AttributeError in _postprocess_batch on the first completed batch. Straight port: NodeSpec carries the innermost enclosing loop, CompiledGraph::loop_order is already Python's _get_loop_order, and num_times_run is on RequestState. A node outside every loop gets the pass index and nothing else, as in Python. Verified against the extension directly (rank 0 -> True, rank 1 -> False; loop context and all three error paths), because base.py is mid-edit on the peek_completion change and the Python suite cannot collect. cargo test: 49 passed. Will re-run the Python parity suite once base.py parses. Co-Authored-By: Claude <noreply@anthropic.com>
…ransport _count_new_tokens took a NodeOutputRouting per rid. Now it takes the flat indices, like the stream-buffer path: numel() still needs the tensors, so the counting stays on this side, but nothing has to carry GraphEdge objects back. Rust had to learn Python's "don't double-count new tokens" for that to be equivalent. One output routed to two destinations is two edges carrying the SAME tensors, so first-edge-of-a-signal-name now wins there; without it the flat indices name each tensor twice and every token count doubles. WorkerGraphsDone.output_signal_names was declared `int` and always carried a list[str]. It survived only because the encoder and decoder for int are both identity, so the list passed through untouched -- and the conductor grew an `isinstance(..., list) else []` guard to swallow the bogus 0 default. Both gone. A Rust encoder written from the annotation would have emitted an int. The Rust GraphRuntime now takes the communicator at construction, like the bookkeeper: PyZmqCommunicator holds its RawZmqCommunicator behind an Arc and hands out a share. No extra lock -- send() takes &self and guards its own sockets. Until now the struct had no transport at all, so worker.py's gate refused MSTAR_RUST_ZMQ=0 on the grounds that the runtime "holds an Arc to the same transport", which was not yet true. test_api_result_delivery stubbed the tensor manager with the pre-refactor `request_id` parameter names. It was the unnamed third failure in every run since the rename; it now passes, leaving only the two known-unrelated ones (qwen3-tts subprocess dep, renamed RaggedPrefillWrapper attribute). Co-Authored-By: Claude <noreply@anthropic.com>
A Rust sender can build the frame it owns and splice Python-owned fields in as opaque values, rather than learning their types. That matters most for resource_publish_info: PublishedInfo is abstract, so every new resource would otherwise become a Rust change. Passing the DECLARED type is what makes the result safe to splice -- it picks the same encoder _plan would have chosen for that field, so the frame decodes identically to having built the whole message in Python. Both tests assert exactly that, against the field as encoded in place. CurrentForwardPassInfo is registered (`fwd_pass_info`), so this never reaches the pickle fallback, which a Rust peer could not decode. Co-Authored-By: Claude <noreply@anthropic.com>
graph/frames.rs encodes the message directly as msgpack, so the highest-volume frame no longer has to be assembled as Python objects and re-encoded per pass. What has to match is what the frame DECODES to, not its bytes -- the receiver runs unpackb and looks fields up by name, so map order and integer width are free. Every test asserts wire.decode(rust_frame) == the message Python would have sent, decoded. Comparing against Python's DECODED message, not the object handed in: the codec is lossy in small real ways. stride is declared list[int] and every caller passes a tuple, so a round trip returns a list -- comparing to the pre-encode object fails Python against itself. Fields whose types belong to Python arrive already encoded and are spliced in untouched: resource_publish_info (PublishedInfo is abstract, so owning it here would mean a Rust change per resource) and the profiling trio, which today's shim decodes only for the outgoing frame to re-encode. rmpv is the vehicle -- re-serialising a generic value is what makes the splice exact. Rust resolves persist descriptors through the bookkeeper it already shares. That needs care: a descriptor's symbols are interned in the BOOKKEEPER's table, not the runtime's, and mixing them up would silently swap one string for another. There is a test for it, and for the omission rule -- shm_segment and the two _source_* fields have defaults, so an explicit nil would decode as a segment literally named nil rather than "spilled to a file". Sends through the shared transport when one was given; returns the bytes either way, which is what makes it testable without a socket. Not yet called from send_outputs -- the shim still builds this message. That swap is next, along with InputSignals and StopLoops. Co-Authored-By: Claude <noreply@anthropic.com>
INPUT_SIGNALS to peer workers and RESULT_TENSORS to the api server join WORKER_GRAPHS_DONE in graph/frames.rs, so send_outputs is now a wrapper that prepares only the Python-owned payloads and hands off. The four per-request buffers moved to RequestInfo.pending. On the request rather than in a side map keyed by handle, deliberately: handles are recycled, and a side map that outlives its request hands one request's persist signals and token counts to whichever request draws that integer next. I patched that leak with explicit pops two commits ago; now remove_request frees them with everything else and the shape cannot come back. CurrentForwardPassInfo is encoded once per request, not per pass -- the worker mutates it exactly once, at admit (rid_handle), and re-sends the same object every pass. resource_publish_info is read back out of that same blob by key rather than encoded again: it is a FIELD of the object Python already sends. Getting that wrong spliced the whole fwd_info into resource_publish_info, which surfaced as `unknown wire tag 'r'` -- a two-character string unpacked as a [tag, payload] pair. Profiling was PICKLED. wire.encode([rx, tx, timings]) takes a bare list, which has no wire tag, so it hit the OPAQUE fallback -- undecodable by a Rust peer and unspliceable into a frame Rust builds, while base.py described it as msgpack. encode_fields encodes each by its declared type; Rust splits the blob into the three fields, and a malformed one yields nothing rather than taking the send down, since profiling is diagnostic. The tests now drive the real path end to end: a completion is routed, send_outputs pushes every frame down the SHARED transport, and the test reads them off a peer's inbox. Nothing is inspected before it has travelled. A bare drain() right after the send races zmq delivery and returns nothing, hence the blocking first read. Co-Authored-By: Claude <noreply@anthropic.com>
The last message type. stop_loops_batched no longer returns a fanout for Python to turn into frames -- it stops the loops, decides who runs them, and sends. loop_stop_times is snapshotted once before the sends rather than read per peer. stop_loops_for_rid has already recorded this stop by then, and every peer has to see the same observation or they cannot agree on which stop is newer. Still never to ourselves: this rank originated the stop and has applied it, and a self-send would land in apply_peer_loop_stops and stop the loops a second time. There is a test for that, and for a loop no peer runs sending nothing at all. _loop_stop_times stays on the Python side -- no longer on a send path, but three test files read it to inspect what a rank observed. Co-Authored-By: Claude <noreply@anthropic.com>
A stalled pipeline looks the same from outside whether the graph never became ready, the scheduler filtered the node out, or the engine keeps saying not-ready. The probe walks those gates in order and names the first one that drops everything. Off unless MSTAR_STALL_PROBE is set to a seconds interval. Co-Authored-By: Claude <noreply@anthropic.com>
Three paths hand back empty outputs for a whole batch without a word: admit refusing the step, _drive_step returning no raw outputs, and _merge_per_rid finding no key for any rid. Downstream cannot tell any of them from a node that genuinely emitted nothing. Routing still runs, every edge carries no tensors, nothing becomes ready, and the pipeline stalls with no error anywhere -- which is exactly the shape of the Orpheus/Bagel stall. The merge case is worth naming precisely: out_ids are the worker's integer rid handles, so a submodule that keyed its output dict by the request id STRING misses every single one. That is the rid refactor's failure mode, and silence made it indistinguishable from an empty node. Only fires when ALL rids miss, so a genuinely partial batch stays quiet. Co-Authored-By: Claude <noreply@anthropic.com>
The stall shows up as RouteInput(tensors=[], num_tensors=[0,...]): the node
runs, every edge carries nothing, nothing downstream becomes ready, and no
error is raised anywhere.
Instrumented at the point the emptiness is observable rather than upstream,
since the three engine paths that return empty outputs all stayed quiet. Two
distinct causes, told apart:
- the rid is absent from the engine's dict, which is a KEY-TYPE mismatch
(these are integer handles, and a submodule keyed by the request id
string misses every one)
- the rid is present and the node emitted names no edge carries, so the
tensors are filed under keys the lookup never asks for
Co-Authored-By: Claude <noreply@anthropic.com>
The all-missed check compared against len(out_ids), but out_ids is the PADDED id list -- so with any padding the count can never reach it and the warning never fires. That is exactly what happened: the batch came through with "present but empty" outputs and this said nothing. Compares against the number of pairs actually zipped now, and adds the case the stall turned out to be: every rid matched a key, and every one of them mapped to an empty dict. That distinguishes "the forward emitted nothing" from "we looked under the wrong keys", which is the whole point of the warning. Co-Authored-By: Claude <noreply@anthropic.com>
A partition only moves to the next walk once every worker graph of the current one has reported from every rank. Stuck there, the request simply stops: the walk never changes, no new inputs go out, and nothing raises -- which is the shape of the Bagel/Orpheus stall. Names the worker graphs still outstanding, how many ranks reported for each, and how many were expected. Quiet for a single-graph partition unless something is actually wrong. Co-Authored-By: Claude <noreply@anthropic.com>
…walk" This reverts commit 86ba5a4.
…match (buggy?) python behavior in streaming_ready (marked FIXME)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What does this PR do?
How was it tested?
Checklist
ruff check .passes