diff --git a/mstar/api_server/data_worker.py b/mstar/api_server/data_worker.py
index da5b80f3af..b262a2f661 100644
--- a/mstar/api_server/data_worker.py
+++ b/mstar/api_server/data_worker.py
@@ -137,7 +137,6 @@ def _kv_page_sizes(model: Model, model_config: dict) -> dict[str, int]:
     }
 
 
-NameToLoopIndices = dict[str, NestedLoopIndices]
 
 
 class PreprocessWorker:
@@ -166,7 +165,8 @@ def __init__(
         self.wakeup = EventWakeup()
 
         self.per_request_reading_tensors = {}
-        self.output_loop_idxs: dict[str, NameToLoopIndices] = {}
+        # request id -> output name -> chunks received, for received_final_chunks
+        self.received_chunk_counts: dict[str, dict[str, int]] = {}
 
         # Build the communicator + tensor manager here (main thread) and hand
         # them to the worker thread, rather than constructing them inside it.
@@ -219,7 +219,7 @@ def _signal(self):
             wakeup.signal()
 
     def new_request(self, input: PreprocessInput):
-        self.output_loop_idxs[input.request_id] = {}
+        self.received_chunk_counts[input.request_id] = {}
         self.per_request_reading_tensors[input.request_id] = 0
         self.request_input_queue.put(input)
         self._signal()
@@ -232,7 +232,7 @@ def abort_request(self, request_id: str):
         # out from under it. Drop only the main-thread bookkeeping here.
         self.abort_request_queue.put(request_id)
         self._signal()
-        self.output_loop_idxs.pop(request_id, None)
+        self.received_chunk_counts.pop(request_id, None)
         self.per_request_reading_tensors.pop(request_id, None)
 
     def finished_reading(self, request_id: str, drained: bool = True):
@@ -244,21 +244,20 @@ def finished_reading(self, request_id: str, drained: bool = True):
         for any in-flight read."""
         self.reads_done_queue.put((request_id, drained))
         self._signal()
-        self.output_loop_idxs.pop(request_id, None)
+        self.received_chunk_counts.pop(request_id, None)
         self.per_request_reading_tensors.pop(request_id, None)
 
     def new_result_tensors(self, input: ResultTensors):
         name = input.graph_edge.name
-        if input.request_id not in self.output_loop_idxs:
+        if input.request_id not in self.received_chunk_counts:
             # Request was removed while this output was still in flight; ack the
             # tensors so the producing worker can reclaim them rather than leak.
             logger.debug("Late result_tensors for cleaned-up request %s, acking and dropping", input.request_id)
             self.discard_result_tensors(input)
             return
 
-        self.output_loop_idxs[input.request_id][name] = input.loop_indices.max(
-            self.output_loop_idxs[input.request_id].get(name, None)
-        )
+        counts = self.received_chunk_counts[input.request_id]
+        counts[name] = counts.get(name, 0) + 1
 
         self.per_request_reading_tensors[input.request_id] += len(input.graph_edge.tensor_info)
         logger.debug(
@@ -277,13 +276,11 @@ def new_result_token(
         transport read and nothing to ack. ``wake=False`` queues without
         ending the thread's idle wait: a frame's caller hands over every
         request's value first and calls ``wake`` once."""
-        if request_id not in self.output_loop_idxs:
+        if request_id not in self.received_chunk_counts:
             logger.debug("Late token for cleaned-up request %s dropped", request_id)
             return
-        if loop_indices is not None:
-            self.output_loop_idxs[request_id][signal] = loop_indices.max(
-                self.output_loop_idxs[request_id].get(signal, None)
-            )
+        counts = self.received_chunk_counts[request_id]
+        counts[signal] = counts.get(signal, 0) + 1
         self.per_request_reading_tensors[request_id] += 1
         self.result_tensor_input_queue.put(
             InlineToken(request_id, value, loop_indices, modality)
@@ -309,18 +306,18 @@ def has_pending_tensors(self, request_id: str):
 
     def received_final_chunks(
         self, request_id: str,
-        final_outputs: dict[str, NestedLoopIndices],
+        output_chunk_counts: dict[str, int],
     ):
         # Every serving walk reports at least one client-facing output, so an
         # empty dict means the walk emitted nothing (or completion raced ahead
         # of every result). Report not-done and let the TTL backstop close the
         # request rather than completing it instantly with zero chunks.
-        if not final_outputs:
+        if not output_chunk_counts:
             return False
+        received = self.received_chunk_counts.get(request_id, {})
         return all(
-            not loop_iters.label_context_gt( # recv'd loop iters is not less than the final_fwd
-                self.output_loop_idxs[request_id].get(name, None)
-            ) for name, loop_iters in final_outputs.items()
+            received.get(name, 0) >= count
+            for name, count in output_chunk_counts.items()
         )
 
     def get_result_chunks(self)-> list[ResultChunk]:
@@ -373,7 +370,7 @@ def get_rx_info(self, request_id: str) -> list[RxInfo]:
     def cleanup_request(self, request_id: str):
         self.cleanup_request_queue.put(request_id)
         self._signal()
-        self.output_loop_idxs.pop(request_id, None)
+        self.received_chunk_counts.pop(request_id, None)
         self.per_request_reading_tensors.pop(request_id, None)
 
     def shutdown(self):
diff --git a/mstar/api_server/entrypoint.py b/mstar/api_server/entrypoint.py
index 0c1f2f1f7b..cf0198a5b7 100644
--- a/mstar/api_server/entrypoint.py
+++ b/mstar/api_server/entrypoint.py
@@ -224,7 +224,7 @@ class PendingRequest:
     profile: RequestProfile
     event: threading.Event = field(default_factory=threading.Event)
     chunks: list[ResultChunk] = field(default_factory=list)
-    final_outputs: dict = field(default_factory=dict)
+    output_chunk_counts: dict[str, int] = field(default_factory=dict)
     consumed_chunks: int = 0
     error: Any | None = None
     error_status: int = 500
@@ -629,7 +629,7 @@ def _prune_recently_completed(self) -> None:
                 req is not None
                 and not self.preprocess_worker.has_pending_tensors(rid)
                 and self.preprocess_worker.received_final_chunks(
-                    rid, req.final_outputs
+                    rid, req.output_chunk_counts
                 )
             )
             if drained or req is None:
@@ -762,14 +762,14 @@ def _process_messages(self) -> None:
                                 logger.info("API server received %s done", rid)
                                 self.recently_completed[rid] = time.time()
 
-                                if not message.body.final_outputs:
+                                if not message.body.output_chunk_counts:
                                     logger.warning(
                                         "Request %s completed with no reported "
                                         "outputs; holding for late results "
                                         "until the cleanup TTL", rid,
                                     )
                                 req = self.pending_requests[rid]
-                                req.final_outputs = message.body.final_outputs
+                                req.output_chunk_counts = message.body.output_chunk_counts
                                 req.profile.timing.conductor_ingest_time = \
                                     message.body.conductor_ingest_time
                                 req.profile.timing.conductor_finish_time = \
diff --git a/mstar/api_server/request_types.py b/mstar/api_server/request_types.py
index 5026004998..2873dcae0e 100644
--- a/mstar/api_server/request_types.py
+++ b/mstar/api_server/request_types.py
@@ -43,10 +43,10 @@ class ResultTokens:
 class RequestComplete:
     """Signals that a request has finished processing."""
     request_id: str
-    # Maps output signal name to its final forward pass number.
-    # The API server waits until all entries are received before
-    # completing the request.
-    final_outputs: dict[str, NestedLoopIndices]
+    # Output signal name -> chunks the workers emitted to the client. The API
+    # server waits until it has received that many of each before completing
+    # the request: completion can overtake the last chunks on the wire.
+    output_chunk_counts: dict[str, int]
     conductor_ingest_time: float
     conductor_finish_time: float
     graph_timings: GraphTimings = field(default_factory=dict)
diff --git a/mstar/conductor/conductor.py b/mstar/conductor/conductor.py
index 6bd8aa87e8..7420a46230 100644
--- a/mstar/conductor/conductor.py
+++ b/mstar/conductor/conductor.py
@@ -28,7 +28,6 @@
 from mstar.distributed.communication import GlobalParallelConfig, WorkerParallelGroups
 from mstar.engine.resources import KVReqConfig, ResourceReqConfig
 from mstar.graph.base import GraphEdge, NodeAndGraphWalk, TensorPointerInfo
-from mstar.graph.loop_indices import NestedLoopIndices
 from mstar.model.base import ForwardPassArgs, Model, WorkerGraph
 from mstar.platforms import apply_process_env_defaults
 from mstar.profile.format import RxInfo, TxInfo
@@ -263,8 +262,9 @@ class RequestData:
     # Per-streaming-connection state, keyed by (from_partition, to_partition, edge_name)
     streaming_connections: dict[tuple[str, str, str], StreamingConnectionState] = field(default_factory=dict)
 
-    # for api server recv bookeeping
-    final_outputs: dict[str, NestedLoopIndices] = field(default_factory=dict)
+    # output name -> chunks the workers emitted to the client; the API server
+    # holds the request open until it has received this many of each
+    output_chunk_counts: dict[str, int] = field(default_factory=dict)
 
     # Conductor-side profiling timestamps, in ``time.perf_counter()`` seconds.
     # Comparable with the api-server stamps because perf_counter is
@@ -1438,7 +1438,7 @@ def _process_request_done(
                 message_type="request_complete",
                 body=RequestComplete(
                     request_id=request_id,
-                    final_outputs=request_data.final_outputs,
+                    output_chunk_counts=request_data.output_chunk_counts,
                     conductor_ingest_time=request_data.conductor_ingest_time,
                     conductor_finish_time=request_data.conductor_finish_time,
                     graph_timings=request_data.graph_timings,
@@ -1499,8 +1499,13 @@ def _process_worker_graphs_done(
             for tx in body.tx_info:
                 request_data.tx_info[(tx.source_entity, tx.edge_name)] = tx
 
+        # From every rank: a rank reports only the chunks it sent itself, and
+        # each of those reaches the API server as its own message.
+        counts = request_data.output_chunk_counts
+        for name in body.output_signal_names:
+            counts[name] = counts.get(name, 0) + 1
+
         pstate = request_data.partition_states.get(partition_name)
-        request_data.final_outputs.update(body.output_loop_indices)
         if pstate is None:
             logger.warning(
                 "WorkerGraphsDone for unknown partition %s (request %s)",
@@ -1538,10 +1543,6 @@ def _process_worker_graphs_done(
                     consumed = body.stream_tokens_consumed.get(conn.edge_name, 0)
                     conn.consumed_count = max(conn.consumed_count, consumed)
 
-            request_data.final_outputs.update(body.output_loop_indices)
-
-            pstate.curr_forward_outputs += body.output_signal_names
-
         # Each wg is only marked complete when all its TP ranks have reported.
         for wg_id in body.worker_graph_ids:
             count = pstate.wg_rank_completions.get(wg_id, 0) + 1
diff --git a/mstar/conductor/request_info.py b/mstar/conductor/request_info.py
index f6baa3f2bc..47c00f55f2 100644
--- a/mstar/conductor/request_info.py
+++ b/mstar/conductor/request_info.py
@@ -122,7 +122,6 @@ class PartitionState:
     # wg_id -> count of distinct TP ranks that have reported completion
     wg_rank_completions: dict[int, int] = field(default_factory=dict)
     num_output_tokens: int = 0
-    curr_forward_outputs: list[str] = field(default_factory=list)
     # resource label -> PublishedInfo, accumulated across all TP ranks and
     # handed back out on the next forward
     resource_publish_info: dict[str, PublishedInfo] = field(default_factory=dict)
diff --git a/mstar/graph/loop_indices.py b/mstar/graph/loop_indices.py
index 2ecce2799a..6860d4bf1e 100644
--- a/mstar/graph/loop_indices.py
+++ b/mstar/graph/loop_indices.py
@@ -38,10 +38,3 @@ def label_context_gt(self, other: "NestedLoopIndices | None", target_loop_name:
             if our_idx < their_idx:
                 return False
         return False
-
-    def max(self, other: "NestedLoopIndices | None") -> "NestedLoopIndices":
-        if other is None:
-            return self
-        if other.label_context_gt(self):
-            return other
-        return self
diff --git a/mstar/graph/runtime/python.py b/mstar/graph/runtime/python.py
index 99fd066a09..0b2a2b8796 100644
--- a/mstar/graph/runtime/python.py
+++ b/mstar/graph/runtime/python.py
@@ -108,10 +108,8 @@ class GraphRuntimeRequestInfo:
     # so a persist signal cannot race the message that announces it.
     pending_persist_signals: list[GraphEdge] = field(default_factory=list)
     pending_new_token_counts: dict[str, dict[str, int]] = field(default_factory=dict)  # by partition
+    # one name per chunk emitted to the client; the conductor counts them
     current_output_chunks: list[str] = field(default_factory=list)
-    output_loop_indices: dict[str, NestedLoopIndices] = field(
-        default_factory=dict
-    )
 
 
 class PythonGraphRuntime(GraphRuntime):
@@ -1509,7 +1507,6 @@ def send_outputs(
                     edge.name for edge in routing.emit_to_client
                 )
                 for edge in routing.emit_to_client:
-                    info.output_loop_indices[edge.name] = nested_idxs.get(rid)
                     self._communicator.send("api_server", APIServerMessage(
                         message_type="result_tensors",
                         body=ResultTensors(
@@ -1618,7 +1615,6 @@ def _send_worker_graphs_done(
                 partition_name=partition,
                 partition_done=partition_done,
                 stream_tokens_consumed=stream_tokens_consumed,
-                output_loop_indices=info.output_loop_indices,
                 graph_timings=graph_timings,
                 rx_info=rx_info,
                 tx_info=tx_info,
diff --git a/mstar/utils/ipc_format.py b/mstar/utils/ipc_format.py
index a8670682b5..20bdb28c4a 100644
--- a/mstar/utils/ipc_format.py
+++ b/mstar/utils/ipc_format.py
@@ -235,12 +235,12 @@ class WorkerGraphsDone(MessageBody):
     is_first_tp_rank: bool
     persist_signals: dict[str, list[TensorPointerInfo]] = field(default_factory=dict)
     new_token_counts: dict[str, int] = field(default_factory=dict) # name to token counts
+    # one name per chunk this worker emitted to the client since its last report
     output_signal_names: list[str] = field(default_factory=list)
     resource_publish_info: dict[str, PublishedInfo] = field(default_factory=dict)
     partition_name: str = field(default="default")
     partition_done: bool = field(default=False)
     stream_tokens_consumed: dict[str, int] = field(default_factory=dict)  # edge_name -> tokens consumed from stream
-    output_loop_indices: dict[str, NestedLoopIndices] = field(default_factory=dict)
     graph_timings: GraphTimings = field(default_factory=dict)
     rx_info: list[RxInfo] = field(default_factory=list)
     tx_info: list[TxInfo] = field(default_factory=list)
diff --git a/mstar/worker/worker.py b/mstar/worker/worker.py
index 73662acf0d..6f9e9931b2 100644
--- a/mstar/worker/worker.py
+++ b/mstar/worker/worker.py
@@ -3245,7 +3245,7 @@ def _pp_stage(name: str) -> None:
             range_push("worker.send_outputs", synchronize=False)
 
         # The consuming pass (not the earlier ingest) reports the partition
-        # done, so it rides this pass's WGD with the final output loop index.
+        # done, so it rides the WGD that also counts this pass's last chunks.
         for rid in batch_N.node_batch.stream_partition_done_rids:
             self._graph_runtime.mark_stream_partition_done(rid, batch_N.partition)
 
diff --git a/rust/src/graph/frames.rs b/rust/src/graph/frames.rs
index 6fe4f3dcac..3853638038 100644
--- a/rust/src/graph/frames.rs
+++ b/rust/src/graph/frames.rs
@@ -239,7 +239,6 @@ pub struct WorkerGraphsDone<'a> {
     pub partition_name: &'a str,
     pub partition_done: bool,
     pub stream_tokens_consumed: &'a [(String, i64)],
-    pub output_loop_indices: Vec<(Sym, (Vec<Sym>, Vec<(Sym, u32)>, u32))>,
     /// `dict[str, PublishedInfo]` -- abstract, so it stays Python's. Read
     /// encoded by Python and forwarded without interpreting resource types.
     pub resource_publish_info: Option<Value>,
@@ -275,14 +274,6 @@ impl WorkerGraphsDone<'_> {
             str_counts(
                 m.key("stream_tokens_consumed"), self.stream_tokens_consumed,
             );
-            w_map(
-                m.key("output_loop_indices"),
-                self.output_loop_indices.len() as u32,
-            );
-            for (sig, (order, idxs, fwd)) in &self.output_loop_indices {
-                w_str(m.buf(), it.name(*sig));
-                nested_loop_indices(m.buf(), it, order, idxs, *fwd);
-            }
             // Omitted when absent, so the decoder's default stands.
             if let Some(v) = &self.resource_publish_info {
                 w_value(m.key("resource_publish_info"), v);
diff --git a/rust/src/graph/request.rs b/rust/src/graph/request.rs
index 7d4a24c18b..dd810e2f79 100644
--- a/rust/src/graph/request.rs
+++ b/rust/src/graph/request.rs
@@ -106,15 +106,14 @@ pub struct PendingOutputs {
     pub persist: Vec<(Sym, Vec<u64>)>,
     /// partition -> signal -> count; each report carries its own partition's.
     pub new_tokens: Vec<(PartitionId, Vec<(String, i64)>)>,
-    /// Emitted signal names, in emission order.
+    /// One name per chunk emitted to the client, in emission order. The
+    /// conductor counts them to tell the API server how many chunks to wait
+    /// for.
     pub output_signals: Vec<Sym>,
-    /// signal -> the loop context it was emitted at.
-    pub output_loop_indices: Vec<(Sym, (Vec<Sym>, Vec<(Sym, u32)>, u32))>,
 }
 
 impl PendingOutputs {
-    /// Everything but `output_loop_indices`, which the conductor keeps for the
-    /// life of the request -- Python re-sends it on every WGD.
+    /// Everything pending, with only `partition`'s token counts.
     pub fn take_for_send(
         &mut self, partition: Option<PartitionId>,
     ) -> (Vec<(Sym, Vec<u64>)>, Vec<(String, i64)>, Vec<Sym>) {
@@ -147,8 +146,7 @@ impl PendingOutputs {
         }
     }
 
-    /// Accumulate, matching Python: a repeated signal's count adds up, and a
-    /// repeated loop index overwrites.
+    /// Accumulate, matching Python: a repeated signal's count adds up.
     pub fn add_new_tokens(&mut self, partition: PartitionId, counts: &[(String, i64)]) {
         let i = match self.new_tokens.iter().position(|(p, _)| *p == partition) {
             Some(i) => i,
@@ -165,15 +163,6 @@ impl PendingOutputs {
             }
         }
     }
-
-    pub fn set_loop_indices(
-        &mut self, signal: Sym, idx: (Vec<Sym>, Vec<(Sym, u32)>, u32),
-    ) {
-        match self.output_loop_indices.iter_mut().find(|(s, _)| *s == signal) {
-            Some(slot) => slot.1 = idx,
-            None => self.output_loop_indices.push((signal, idx)),
-        }
-    }
 }
 
 impl RequestInfo {
diff --git a/rust/src/graph/runtime.rs b/rust/src/graph/runtime.rs
index 01ed956b39..8a7c7b5510 100644
--- a/rust/src/graph/runtime.rs
+++ b/rust/src/graph/runtime.rs
@@ -320,9 +320,6 @@ pub struct GraphRuntime {
     /// signals, token counts and emitted signal names -- they accumulate
     /// across sends and flush onto the next completion, so a persist signal
     /// cannot race the message announcing it.
-    ///
-    /// `output_loop_indices` is NOT drained: the conductor keeps it for the
-    /// life of the request, and Python re-sends it on every WGD.
     fn worker_graphs_done_frame(
         &mut self,
         rid: u32,
@@ -346,13 +343,10 @@ pub struct GraphRuntime {
                 .unwrap_or(false);
 
         let partition = self.interner.get(partition_name);
-        let (persist, new_tokens, output_signals, loop_indices) =
+        let (persist, new_tokens, output_signals) =
             match self.requests[rid as usize].as_mut() {
-                Some(info) => {
-                    let (p, n, o) = info.pending.take_for_send(partition);
-                    (p, n, o, info.pending.output_loop_indices.clone())
-                }
-                None => (vec![], vec![], vec![], vec![]),
+                Some(info) => info.pending.take_for_send(partition),
+                None => (vec![], vec![], vec![]),
             };
         let persist_signals: Vec<(Sym, Vec<crate::tensors::TensorPointerInfo>)> =
             persist.into_iter().map(|(sig, uuids)| (sig, self.infos(&uuids))).collect();
@@ -367,7 +361,6 @@ pub struct GraphRuntime {
             partition_name,
             partition_done,
             stream_tokens_consumed,
-            output_loop_indices: loop_indices,
             resource_publish_info: frames::spliced_value(publication_encoded),
             profiling: frames::split_profiling(profiling_encoded),
         };
@@ -3302,9 +3295,6 @@ impl GraphRuntime {
                 let loop_indices = nested.get(&rid).cloned();
                 if let Some(info) = self.requests[rid as usize].as_mut() {
                     info.pending.output_signals.push(sig);
-                    if let Some(idx) = loop_indices.clone() {
-                        info.pending.set_loop_indices(sig, idx);
-                    }
                 }
                 let bytes = {
                     let bk = self.bookkeeping.lock().unwrap();
@@ -3322,8 +3312,8 @@ impl GraphRuntime {
 
             // Inline emits: one RESULT_TOKENS frame per (signal, modality)
             // per step, with the same per-request bookkeeping as a tensor
-            // emit so the completion report names the signal and its loop
-            // context. Usually one group.
+            // emit so the completion report counts the chunk. Usually one
+            // group.
             if !plan.emit_inline.is_empty() {
                 let mut groups: Vec<((String, String), Vec<(u32, i64)>)> = Vec::new();
                 for (rid, signal, modality, value) in plan.emit_inline {
@@ -3343,9 +3333,6 @@ impl GraphRuntime {
                         let idx = nested.get(&rid).cloned();
                         if let Some(info) = self.requests[rid as usize].as_mut() {
                             info.pending.output_signals.push(sig);
-                            if let Some(i) = idx.clone() {
-                                info.pending.set_loop_indices(sig, i);
-                            }
                         }
                         request_ids.push(request_id);
                         values.push(value);
diff --git a/test/modular/test_api_completion_guard.py b/test/modular/test_api_completion_guard.py
index 0b010e07ca..8cba57be8f 100644
--- a/test/modular/test_api_completion_guard.py
+++ b/test/modular/test_api_completion_guard.py
@@ -2,7 +2,7 @@
 fully delivered, and a request that fails preprocessing must reach the
 client as an error instead of leaving it to hit the server timeout.
 
-An empty final_outputs dict made received_final_chunks vacuously true, so a
+An empty output_chunk_counts dict made received_final_chunks vacuously true, so a
 request whose walk emitted nothing (or whose completion message raced ahead
 of every result) completed instantly with zero chunks instead of holding for
 late results.
@@ -14,29 +14,25 @@
 
 from mstar.api_server.data_worker import PreprocessWorker, PreprocessWorkerThread
 from mstar.api_server.request_types import PreprocessInput
-from mstar.graph.loop_indices import NestedLoopIndices
 
 
-def _worker_with(output_loop_idxs):
+def _worker_with(received_chunk_counts):
     worker = PreprocessWorker.__new__(PreprocessWorker)
-    worker.output_loop_idxs = output_loop_idxs
+    worker.received_chunk_counts = received_chunk_counts
     return worker
 
 
-def test_empty_final_outputs_is_not_done():
+def test_empty_output_chunk_counts_is_not_done():
     worker = _worker_with({"r1": {}})
     assert worker.received_final_chunks("r1", {}) is False
 
 
-def test_final_outputs_wait_for_registration_then_complete():
-    final = NestedLoopIndices(
-        loop_name_order=["denoise"], loop_indices={"denoise": 34}, wg_fwd_pass_idx=0
-    )
+def test_output_chunk_counts_wait_for_registration_then_complete():
     worker = _worker_with({"r1": {}})
     # Completion reported but the terminal chunk hasn't registered yet.
-    assert worker.received_final_chunks("r1", {"video_output": final}) is False
-    worker.output_loop_idxs["r1"]["video_output"] = final
-    assert worker.received_final_chunks("r1", {"video_output": final}) is True
+    assert worker.received_final_chunks("r1", {"video_output": 1}) is False
+    worker.received_chunk_counts["r1"]["video_output"] = 1
+    assert worker.received_final_chunks("r1", {"video_output": 1}) is True
 
 
 def test_preprocess_failure_emits_error_chunk():
diff --git a/test/modular/test_api_result_delivery.py b/test/modular/test_api_result_delivery.py
index 74bd4ae8a0..a01c81a358 100644
--- a/test/modular/test_api_result_delivery.py
+++ b/test/modular/test_api_result_delivery.py
@@ -141,7 +141,7 @@ def __init__(self, pending=False, final=True):
     def has_pending_tensors(self, rid):
         return self.pending
 
-    def received_final_chunks(self, rid, final_outputs):
+    def received_final_chunks(self, rid, output_chunk_counts):
         return self.final
 
     def finished_reading(self, rid, drained=True):
diff --git a/test/modular/test_cfg_tp_publish.py b/test/modular/test_cfg_tp_publish.py
index d68b593f89..f1ca1b04ce 100644
--- a/test/modular/test_cfg_tp_publish.py
+++ b/test/modular/test_cfg_tp_publish.py
@@ -50,7 +50,7 @@ def test_conductor_merges_kv_publish_info_from_every_tp_rank():
     )
     request = SimpleNamespace(
         partition_states={"default": partition},
-        final_outputs={},
+        output_chunk_counts={},
         persist_signals={},
         streaming_connections={},
         worker_graph_to_workers={"wg": ["worker_0", "worker_1"]},
diff --git a/test/modular/test_data_worker_late_release.py b/test/modular/test_data_worker_late_release.py
index b6800374f5..483fc65bf8 100644
--- a/test/modular/test_data_worker_late_release.py
+++ b/test/modular/test_data_worker_late_release.py
@@ -143,7 +143,7 @@ def __init__(self, model, start=True):
         facade.result_tensor_input_queue = self.worker.result_tensor_queue
         facade.abort_request_queue = self.worker.abort_request_queue
         facade.reads_done_queue = self.worker.reads_done_queue
-        facade.output_loop_idxs = {}
+        facade.received_chunk_counts = {}
         facade.per_request_reading_tensors = {}
         self.api = facade
         self.thread = threading.Thread(target=self.worker.run, daemon=True)
diff --git a/test/modular/test_delivery_chunk_counts.py b/test/modular/test_delivery_chunk_counts.py
new file mode 100644
index 0000000000..96714490f0
--- /dev/null
+++ b/test/modular/test_delivery_chunk_counts.py
@@ -0,0 +1,349 @@
+"""The API server holds a finished request open until every chunk the workers
+emitted has arrived.
+
+``request_complete`` (worker -> conductor -> API server) and a worker's last
+result chunks (worker -> API server) travel on different connections, so
+completion can overtake them. The conductor counts the chunks each worker
+reports emitting (``WorkerGraphsDone.output_signal_names``, one name per
+chunk, from every rank), and the API server compares that with what it has
+received.
+
+These drive the real Python graph runtime, the conductor's report handling and
+the API server's bookkeeping; ``test/rust/test_delivery_chunk_counts_rust.py``
+runs the same request on the Rust runtime.
+"""
+from types import SimpleNamespace
+
+import pytest
+import torch
+
+from mstar.api_server.data_worker import PreprocessWorker
+from mstar.api_server.request_types import ResultTensors, ResultTokens
+from mstar.communication.tensor_store import TensorStore
+from mstar.communication.tensors import TensorCommunicationManager
+from mstar.conductor.conductor import Conductor, RequestData
+from mstar.conductor.request_info import CurrentForwardConductorMetadata, PartitionState
+from mstar.distributed.base import ShardingConfig
+from mstar.graph.base import GraphEdge, GraphNode, Loop, TensorPointerInfo
+from mstar.graph.runtime.base import ColumnarEdgeSpecs, RouteInput, SendInput
+from mstar.graph.runtime.python import PythonGraphRuntime
+from mstar.graph.special_destinations import EMIT_TO_CLIENT
+from mstar.model.base import WorkerGraph
+from mstar.utils.containers import ParallelList
+from mstar.utils.ipc_format import ConductorMessageType, WorkerGraphsDone
+
+WORKER = "worker_0"
+PREFILL, DECODE, LOOP = "prefill_text", "thinker_decode", "thinker_decode_loop"
+NODE = "Thinker"
+
+# (prefill runs, decode steps): Qwen3-Omni's "text, image, text" prompt runs
+# prefill_text twice (the image walk emits on its own worker graph, which
+# changes nothing here); one prefill with one decode step is any AR model whose
+# decode stops after a token (EOS, or max_output_tokens=2).
+SHAPES = {
+    "text_image_text": (2, 3),
+    "one_decode_step": (1, 1),
+    "plain": (1, 3),
+}
+
+
+def worker_graphs(decode_steps: int) -> list[WorkerGraph]:
+    prefill = GraphNode(name=NODE, input_names=["text_inputs"], outputs=[
+        GraphEdge(next_node=EMIT_TO_CLIENT, name="new_token", output_modality="text", persist=True),
+    ])
+    decode = Loop(name=LOOP, max_iters=decode_steps, outputs=[], section=GraphNode(
+        name=NODE, input_names=["text_inputs"], outputs=[
+            GraphEdge(next_node=EMIT_TO_CLIENT, name="new_token", output_modality="text"),
+            GraphEdge(next_node=NODE, name="text_inputs"),
+        ],
+    ))
+    return [
+        WorkerGraph(section=prefill, graph_walks={PREFILL}, ranks=[0], worker_graph_id=0),
+        WorkerGraph(section=decode, graph_walks={DECODE}, ranks=[0], worker_graph_id=1),
+    ]
+
+
+RUNTIME_KWARGS = dict(
+    my_worker_id=WORKER,
+    all_wg_ids_to_graph_walks={0: {PREFILL}, 1: {DECODE}},
+    all_wg_ids_to_dyn_loops={0: set(), 1: {LOOP}},
+    all_wg_ids_to_nodes={0: {NODE}, 1: {NODE}},
+    node_to_partition={NODE: "default"},
+    sharding_config=ShardingConfig(groups=[], tp_enabled_nodes=set(), shard_dim={}),
+)
+
+
+def tensor_info(uuid: int) -> TensorPointerInfo:
+    return TensorPointerInfo(
+        dims=[1], dtype=torch.int64, nbytes=8, address=0, stride=[1], uuid=uuid,
+        source_session_id="h:1", source_entity=WORKER,
+    )
+
+
+def drive(runtime, put, prefills: int, decode_steps: int, inline: bool = False) -> None:
+    """Run one request through ``prefills`` prefill walks and a decode loop.
+
+    ``put(uuid)`` registers a produced tensor with the runtime's store, as the
+    worker's prologue does. ``inline`` sends the decode tokens as inline values
+    (the Rust runtime's RESULT_TOKENS path).
+    """
+    rid = runtime.add_request(
+        request_id="r1", partition="default", graph_walk=PREFILL,
+        partition_worker_graph_ids=[0, 1],
+        worker_graph_to_workers=ParallelList([0, 1], [[WORKER], [WORKER]]),
+    )
+    uuid = 100
+
+    def step(walk: str, wg_id: int, inline_value=None):
+        nonlocal uuid
+        assert runtime.pop_rids(NODE, walk, [rid]) is not None
+        signals = runtime.get_output_signals(NODE, walk)
+        uuids = []
+        for _ in signals:
+            uuid += 1
+            put(uuid)
+            uuids.append(uuid)
+        extra = {} if inline_value is None else dict(
+            inline_signal="new_token", inline_values=[inline_value],
+        )
+        out = runtime.complete_and_route_batch(RouteInput(
+            partition="default", graph_walk=walk, node_name=NODE, output_signals=signals,
+            wg_ids=ParallelList([rid], [wg_id]), tensors=uuids,
+            num_tensors=[1] * len(signals), **extra,
+        ))
+        runtime.send_outputs(SendInput(
+            completion_id=out.completion_id, per_request_info=ParallelList([rid], [None]),
+            new_token_counts=ParallelList([rid], [{}]),
+        ))
+
+    def ingest():
+        block = ColumnarEdgeSpecs.empty()
+        block.add_edge(rid, GraphEdge(name="text_inputs", next_node=NODE))
+        assert runtime.ingest_inputs_batch(block, can_buffer=True) == []
+
+    for _ in range(prefills):
+        runtime.set_walk(rid, "default", PREFILL)
+        ingest()
+        step(PREFILL, 0)
+    runtime.set_walk(rid, "default", DECODE)
+    ingest()
+    for i in range(decode_steps):
+        step(DECODE, 1, inline_value=7 + i if inline else None)
+
+
+def expected_counts(reports: list[WorkerGraphsDone]) -> dict[str, int]:
+    """What the conductor puts in ``request_complete``, from the real handler."""
+    conductor = Conductor.__new__(Conductor)
+    conductor.enable_prof = False
+    request = RequestData(
+        persist_signals={}, persist_signal_ref_cnt={},
+        worker_graph_to_workers={0: [WORKER], 1: [WORKER]}, all_worker_graph_ids={0, 1},
+        max_output_tokens=64, random_seed=0, resource_configs={},
+        partition_states={"default": PartitionState(
+            partition_name="default",
+            metadata=CurrentForwardConductorMetadata(graph_walk=PREFILL, is_prefill=True),
+        )},
+    )
+    conductor.requests = {"r1": request}
+    for body in reports:
+        conductor._process_worker_graphs_done(body)
+    return request.output_chunk_counts
+
+
+def api_server() -> PreprocessWorker:
+    worker = PreprocessWorker.__new__(PreprocessWorker)
+    worker.received_chunk_counts = {"r1": {}}
+    worker.per_request_reading_tensors = {"r1": 0}
+    worker.result_tensor_input_queue = SimpleNamespace(put=lambda item: None)
+    worker._signal = lambda: None
+    return worker
+
+
+def deliver(worker: PreprocessWorker, chunk) -> None:
+    """One chunk message as the API server's message loop hands it over."""
+    if isinstance(chunk, ResultTokens):
+        for i, rid in enumerate(chunk.request_ids):
+            worker.new_result_token(
+                rid, chunk.values[i], chunk.loop_indices[i], chunk.signal, chunk.modality,
+            )
+    else:
+        assert isinstance(chunk, ResultTensors)
+        worker.new_result_tensors(chunk)
+
+
+def assert_holds_until_last_chunk(chunks: list, reports: list[WorkerGraphsDone]) -> None:
+    counts = expected_counts(reports)
+    assert counts == {"new_token": len(chunks)}
+    worker = api_server()
+    for i, chunk in enumerate(chunks):
+        assert not worker.received_final_chunks("r1", counts), (
+            f"request reported delivered with {i} of {len(chunks)} chunks received"
+        )
+        deliver(worker, chunk)
+    assert worker.received_final_chunks("r1", counts)
+
+
+class _Recorder:
+    def __init__(self):
+        self.sent = []
+
+    def send(self, entity_id, msg):
+        self.sent.append((entity_id, msg))
+
+    def chunks(self):
+        return [m.body for e, m in self.sent if e == "api_server"]
+
+    def reports(self):
+        return [m.body for e, m in self.sent
+                if e == "conductor" and m.message_type == ConductorMessageType.WORKER_GRAPHS_DONE]
+
+
+class _TensorManager:
+    """What the Python runtime needs of the manager: a real store underneath."""
+
+    stamps_shm_placement = False
+    refresh_shm_placement = TensorCommunicationManager.refresh_shm_placement
+    set_output_ref_counts = TensorCommunicationManager.set_output_ref_counts
+
+    def __init__(self):
+        self.tensor_store = TensorStore()
+
+    def set_persist(self, uuid, persist):
+        self.tensor_store.set_metadata(uuid, persist=persist)
+
+    def increment_ref(self, uuid, n=1):
+        self.tensor_store.increment_ref(uuid, n=n)
+
+    def dereference_batch(self, uuids, counts):
+        self.tensor_store.dereference_batch(uuids, counts)
+
+    def dereference_batch_uniform(self, uuids, n=1):
+        pass
+
+
+def python_runtime(decode_steps: int):
+    manager = _TensorManager()
+    recorder = _Recorder()
+    runtime = PythonGraphRuntime(
+        my_worker_graphs=worker_graphs(decode_steps), tensor_manager=manager,
+        communicator=recorder, **RUNTIME_KWARGS,
+    )
+
+    def put(uuid):
+        manager.tensor_store.put_tensor("r1", uuid, torch.ones(1), tensor_info(uuid))
+
+    return runtime, put, recorder
+
+
+@pytest.mark.parametrize("shape", list(SHAPES))
+def test_request_is_held_until_its_last_chunk(shape):
+    prefills, decode_steps = SHAPES[shape]
+    runtime, put, recorder = python_runtime(decode_steps)
+    drive(runtime, put, prefills, decode_steps)
+    assert len(recorder.chunks()) == prefills + decode_steps
+    assert_holds_until_last_chunk(recorder.chunks(), recorder.reports())
+
+
+class _ScriptedApiComm:
+    """The API server's inbox: one scripted batch per loop pass, stopping the
+    loop when the script runs out. Records whether the client had been
+    released before each pass."""
+
+    def __init__(self, server, batches, req):
+        self.server, self.batches, self.req = server, list(batches), req
+        self.released_before = []
+
+    def get_all_new_messages(self):
+        self.released_before.append(self.req.event.is_set())
+        if not self.batches:
+            self.server.running = False
+            return []
+        return self.batches.pop(0)
+
+
+def test_completion_that_overtakes_the_last_chunks_waits_for_them():
+    """Through the API server's real message loop: ``request_complete``
+    arrives after two of the five chunks, the rest a pass later each. The
+    client is released only once the fifth has been read, never with a
+    truncated reply."""
+    import collections
+    import queue
+    import threading
+
+    from mstar.api_server.entrypoint import APIServer, PendingRequest
+    from mstar.api_server.request_types import APIServerMessage, RequestComplete, ResultChunk
+    from mstar.profile.format import RequestProfile
+
+    runtime, put, recorder = python_runtime(SHAPES["text_image_text"][1])
+    drive(runtime, put, *SHAPES["text_image_text"])
+    chunks = recorder.chunks()
+    complete = APIServerMessage(message_type="request_complete", body=RequestComplete(
+        request_id="r1", output_chunk_counts=expected_counts(recorder.reports()),
+        conductor_ingest_time=0.0, conductor_finish_time=0.0,
+    ))
+
+    worker = PreprocessWorker.__new__(PreprocessWorker)
+    worker.received_chunk_counts = {"r1": {}}
+    worker.per_request_reading_tensors = {"r1": 0}
+    worker.output_queue = queue.Queue()
+    worker.profile_queue = queue.Queue()
+    worker.reads_done_queue = queue.Queue()
+    worker._signal = lambda: None
+    # the data worker's thread, collapsed: a queued read completes at once
+    worker.result_tensor_input_queue = SimpleNamespace(put=lambda item: worker.output_queue.put(
+        ResultChunk(request_id=item.request_id, modality="text", data=b"x", metadata={}),
+    ))
+
+    server = object.__new__(APIServer)
+    server.running = True
+    server._liveness_interval_s = 3600.0
+    server._conductor_exited = lambda: None
+    server.request_lock = threading.Lock()
+    server.recently_completed = collections.OrderedDict()
+    server._recently_completed_ttl = 3600.0
+    server.log_stats = False
+    server.enable_nvtx = False
+    server.preprocess_worker = worker
+    req = PendingRequest(streaming=False, input_modalities=["text", "image", "text"],
+                         output_modalities=["text"], profile=RequestProfile(rid="r1"))
+    server.pending_requests = {"r1": req}
+
+    def msg(body):
+        return APIServerMessage(message_type="result_tensors", body=body)
+
+    batches = [[msg(chunks[0]), msg(chunks[1])], [complete], []] + [[msg(c)] for c in chunks[2:]] + [[]]
+    server.communicator = _ScriptedApiComm(server, batches, req)
+    server._process_messages()
+
+    assert req.event.is_set() and req.error is None
+    assert len(req.chunks) == len(chunks)
+    # Not released until the pass after the one that read the last chunk (the
+    # loop prunes at the top of a pass); before the fix it was released right
+    # after request_complete.
+    released = server.communicator.released_before
+    last_chunk_pass = len(batches) - 2
+    assert released.index(True) == last_chunk_pass + 1, released
+
+
+def test_every_rank_reports_the_chunks_it_sent():
+    """Under TP a replicated output leaves from rank 0 only and a sharded one
+    from every rank, each rank's copy its own message, so the conductor counts
+    every rank's report rather than only rank 0's."""
+    conductor = Conductor.__new__(Conductor)
+    conductor.enable_prof = False
+    request = RequestData(
+        persist_signals={}, persist_signal_ref_cnt={},
+        worker_graph_to_workers={0: ["worker_0", "worker_1"]}, all_worker_graph_ids={0},
+        max_output_tokens=64, random_seed=0, resource_configs={},
+        partition_states={"default": PartitionState(
+            partition_name="default",
+            metadata=CurrentForwardConductorMetadata(graph_walk=DECODE, is_prefill=False),
+        )},
+    )
+    conductor.requests = {"r1": request}
+    for first, names in [(True, ["sharded", "replicated"]), (False, ["sharded"])]:
+        conductor._process_worker_graphs_done(WorkerGraphsDone(
+            request_id="r1", worker_graph_ids=[0], is_first_tp_rank=first,
+            output_signal_names=names,
+        ))
+    assert request.output_chunk_counts == {"sharded": 2, "replicated": 1}
diff --git a/test/modular/test_gc_freeze.py b/test/modular/test_gc_freeze.py
index cdff3d083a..339e306d82 100644
--- a/test/modular/test_gc_freeze.py
+++ b/test/modular/test_gc_freeze.py
@@ -24,7 +24,7 @@ def test_freeze_after_setup_moves_the_heap():
 
 def test_a_frame_of_tokens_wakes_the_thread_once():
     facade = PreprocessWorker.__new__(PreprocessWorker)
-    facade.output_loop_idxs = {"r1": {}, "r2": {}}
+    facade.received_chunk_counts = {"r1": {}, "r2": {}}
     facade.per_request_reading_tensors = {"r1": 0, "r2": 0}
 
     class _Q:
diff --git a/test/modular/test_inline_tokens.py b/test/modular/test_inline_tokens.py
index 566322ab55..388aabf95e 100644
--- a/test/modular/test_inline_tokens.py
+++ b/test/modular/test_inline_tokens.py
@@ -130,12 +130,12 @@ def test_the_facade_accounts_an_inline_token_like_a_tensor():
     facade = PreprocessWorker.__new__(PreprocessWorker)
     facade.result_tensor_input_queue = queue.Queue()
     facade.per_request_reading_tensors = {"r1": 0}
-    facade.output_loop_idxs = {"r1": {}}
+    facade.received_chunk_counts = {"r1": {}}
     facade.new_result_token("r1", 42, _loop(4), "text_inputs", "text")
     facade.new_result_token("r1", 43, _loop(5), "text_inputs", "text")
     facade.new_result_token("unknown", 1, _loop(0), "text_inputs", "text")  # dropped, no raise
     assert facade.per_request_reading_tensors["r1"] == 2
-    assert facade.output_loop_idxs["r1"]["text_inputs"].loop_indices == {"decode_loop": 5}
+    assert facade.received_chunk_counts["r1"] == {"text_inputs": 2}
     items = [facade.result_tensor_input_queue.get_nowait() for _ in range(2)]
     assert [i.value for i in items] == [42, 43]
     assert facade.result_tensor_input_queue.empty()
diff --git a/test/modular/test_request_id_reuse.py b/test/modular/test_request_id_reuse.py
index 96dd23d321..2ff3fa24a6 100644
--- a/test/modular/test_request_id_reuse.py
+++ b/test/modular/test_request_id_reuse.py
@@ -138,7 +138,8 @@ def __init__(self):
         facade.reads_done_queue = self.worker.reads_done_queue
         facade.cleanup_request_queue = self.worker.cleanup_request_queue
         facade.discard_tensor_queue = self.worker.discard_tensor_queue
-        facade.output_loop_idxs = {}
+        facade.output_queue = self.worker.out_queue
+        facade.received_chunk_counts = {}
         facade.per_request_reading_tensors = {}
         self.data_worker = facade
         self.server = _api_server(facade)
@@ -254,6 +255,80 @@ def test_the_older_request_s_remove_does_not_unlink_the_reuse_s_inputs(rig):
     assert chunk is not None and chunk.data == b"<7>"
 
 
+def _received(rig, key):
+    """A worker's chunk for ``key`` as the message thread hands it over; returns
+    the chunks the API server reads back from the data worker."""
+    rig.data_worker.new_result_token(key, 7, None, "text_inputs", "text")
+    rig.settle()
+    return rig.data_worker.get_result_chunks()
+
+
+def _completed(s, key, output_chunk_counts):
+    """The conductor's request_complete for ``key``, then one prune pass;
+    returns whether the request was released."""
+    with s.request_lock:
+        s.recently_completed[key] = time.time()
+        s.pending_requests[key].output_chunk_counts = output_chunk_counts
+        s._prune_recently_completed()
+        return s.pending_requests[key].event.is_set()
+
+
+def test_a_reuse_does_not_inherit_the_older_request_s_chunks(rig):
+    """X finished with two chunks reported and one received, so it is held open
+    and its id stays in use. The client aborts it and resubmits X; the older
+    request's last chunk lands after the reuse started and must not count
+    toward the reuse's hold."""
+    s = rig.server
+    first = rig.submit("first").request_id
+    assert len(_received(rig, first)) == 1
+    assert not _completed(s, first, {"text_inputs": 2})
+    with pytest.raises(HTTPException) as refused:
+        s.submit_request(text="early", request_id="X", streaming=False, **TEXT)
+    assert refused.value.status_code == 409
+
+    s.abort_request("X")
+    rig.settle()
+    second = rig.submit("second").request_id
+    assert _received(rig, first) == []
+
+    assert not _completed(s, second, {"text_inputs": 1}), (
+        "the reuse was released on the older request's chunk"
+    )
+    assert [c.request_id for c in _received(rig, second)] == [second]
+    with s.request_lock:
+        s._prune_recently_completed()
+    assert s.pending_requests[second].event.is_set()
+
+
+def test_the_older_request_s_release_does_not_release_the_reuse(rig):
+    """X failed, which frees the id while the data worker still holds its
+    state until the client lets go. The reuse is held for its second chunk when
+    the older request is released; the reuse stays held until that chunk."""
+    s = rig.server
+    first = rig.submit("first").request_id
+    with s.request_lock:
+        s.pending_requests[first].event.set()
+        s.recently_completed[first] = time.time()
+
+    second = rig.submit("second").request_id
+    assert len(_received(rig, second)) == 1
+    assert not _completed(s, second, {"text_inputs": 2})
+
+    with s.request_lock:
+        s.pending_requests.pop(first)  # its client read the error
+        s._prune_recently_completed()
+    assert first not in s.recently_completed
+    assert not s.pending_requests[second].event.is_set(), (
+        "the older request's release released the reuse"
+    )
+    assert rig.data_worker.received_chunk_counts[second] == {"text_inputs": 1}
+
+    assert len(_received(rig, second)) == 1
+    with s.request_lock:
+        s._prune_recently_completed()
+    assert s.pending_requests[second].event.is_set()
+
+
 def test_the_seed_still_derives_from_the_client_request_id(rig):
     from mstar.conductor.conductor import _req_id_to_seed, _request_seed, _seed_id
 
diff --git a/test/modular/test_wire_codec.py b/test/modular/test_wire_codec.py
index ec693b2e3d..984ec5ae34 100644
--- a/test/modular/test_wire_codec.py
+++ b/test/modular/test_wire_codec.py
@@ -17,7 +17,7 @@
 import torch
 
 import mstar.communication.wire_types  # noqa: F401  (registers the tags)
-from mstar.api_server.request_types import APIServerMessage, ResultTensors
+from mstar.api_server.request_types import APIServerMessage, RequestComplete, ResultTensors
 from mstar.communication.wire import _TYPE_TO_TAG, WireError, decode, encode
 from mstar.conductor.request_info import CurrentForwardPassInfo
 from mstar.graph.base import GraphEdge, TensorPointerInfo
@@ -200,9 +200,7 @@ def test_worker_graphs_done_full():
             resource_publish_info={},
             partition_name="p0", partition_done=True,
             stream_tokens_consumed={"audio": 12},
-            output_loop_indices={"tok": NestedLoopIndices(
-                loop_name_order=["decode_loop"], loop_indices={"decode_loop": 9},
-                wg_fwd_pass_idx=1)},
+            output_signal_names=["tok", "tok"],
             rx_info=[RxInfo(edge_name="e", source_entity="w1", dest_entity="w0",
                             count=2, num_bytes=99, time=0.5)],
             tx_info=[TxInfo(edge_name="e", source_entity="w0", count=1,
@@ -218,6 +216,19 @@ def test_worker_graphs_done_full():
     assert body.rx_info[0].num_bytes == 99
     assert body.tx_info[0].time == pytest.approx(0.25)
     assert body.partition_done is True
+    assert body.output_signal_names == ["tok", "tok"]
+
+
+def test_api_server_request_complete_carries_chunk_counts():
+    msg = APIServerMessage(
+        message_type="request_complete",
+        body=RequestComplete(
+            request_id="r0", output_chunk_counts={"new_token": 5, "audio_chunk": 2},
+            conductor_ingest_time=1.0, conductor_finish_time=2.0,
+        ),
+    )
+    body = roundtrip(msg).body
+    assert body.output_chunk_counts == {"new_token": 5, "audio_chunk": 2}
 
 
 def test_api_server_result_tensors():
diff --git a/test/rust/test_delivery_chunk_counts_rust.py b/test/rust/test_delivery_chunk_counts_rust.py
new file mode 100644
index 0000000000..6e8a22bf6a
--- /dev/null
+++ b/test/rust/test_delivery_chunk_counts_rust.py
@@ -0,0 +1,74 @@
+"""``test/modular/test_delivery_chunk_counts.py`` on the Rust runtime: the
+frames it actually sends (read off real inboxes) must let the conductor count
+every chunk, so the API server holds the request until the last one arrives.
+Includes the inline-token path, which only the Rust runtime takes.
+"""
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+sys.path.insert(0, ".")
+# the shared scenario lives beside the Python-runtime test; "test" itself can
+# resolve to the standard library's package, so import it by directory
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "modular"))
+
+import pytest
+
+pytest.importorskip("mstar_rust")
+
+from mstar_rust import ZmqCommunicator
+from test_delivery_chunk_counts import (  # noqa: E402
+    RUNTIME_KWARGS,
+    SHAPES,
+    WORKER,
+    assert_holds_until_last_chunk,
+    drive,
+    tensor_info,
+    worker_graphs,
+)
+
+import mstar.communication.wire_types  # noqa: F401  (registers the tags)
+from mstar.api_server.request_types import ResultTokens
+from mstar.communication.tensor_store import RustTensorBookkeeping
+from mstar.communication.wire import decode
+from mstar.graph.runtime import rust as rust_runtime
+from mstar.utils.ipc_format import ConductorMessageType
+
+
+def _collect(box, first_wait_ms=1000):
+    out = []
+    first = box.recv_timeout(first_wait_ms)
+    if first is None:
+        return out
+    out.append(decode(first))
+    while (more := box.recv_timeout(50)) is not None:
+        out.append(decode(more))
+    return out
+
+
+@pytest.mark.parametrize("inline", [False, True], ids=["tensor", "inline"])
+@pytest.mark.parametrize("shape", list(SHAPES))
+def test_request_is_held_until_its_last_chunk(shape, inline, tmp_path):
+    prefills, decode_steps = SHAPES[shape]
+    inboxes = {peer: ZmqCommunicator(peer, str(tmp_path)) for peer in ("conductor", "api_server")}
+    book = RustTensorBookkeeping()
+    runtime = rust_runtime.RustGraphRuntime(
+        my_worker_graphs=worker_graphs(decode_steps), bookkeeping=book,
+        communicator=SimpleNamespace(_inner=ZmqCommunicator(WORKER, str(tmp_path))),
+        **RUNTIME_KWARGS,
+    )
+
+    def put(uuid):
+        book.put_tensor(uuid, tensor_info(uuid))
+        book.increment_ref(uuid, 1)  # the worker's safety hold
+
+    drive(runtime, put, prefills, decode_steps, inline=inline)
+    chunks = [m.body for m in _collect(inboxes["api_server"])]
+    reports = [
+        m.body for m in _collect(inboxes["conductor"])
+        if m.message_type == ConductorMessageType.WORKER_GRAPHS_DONE
+    ]
+    assert len(chunks) == prefills + decode_steps
+    tokens = sum(isinstance(c, ResultTokens) for c in chunks)
+    assert tokens == (decode_steps if inline else 0)
+    assert_holds_until_last_chunk(chunks, reports)
diff --git a/test/rust/test_rust_wire_frames.py b/test/rust/test_rust_wire_frames.py
index 3026b04280..72682ab9a7 100644
--- a/test/rust/test_rust_wire_frames.py
+++ b/test/rust/test_rust_wire_frames.py
@@ -227,10 +227,9 @@ def test_an_emitted_signal_reaches_the_api_server(tmp_path):
             loop_indices=idx, metadata={},
         ),
     ))]
-    # The WGD reports what was emitted, and at which loop index.
+    # The WGD names one entry per emitted chunk, for the conductor to count.
     wgd = got["conductor"][0].body
     assert wgd.output_signal_names == ["out"]
-    assert wgd.output_loop_indices == {"out": idx}
 
 
 def test_a_remote_destination_gets_input_signals(tmp_path):