fix: stabilize session actor fanout (#2062)

Co-authored-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com>
Co-authored-by: nacho.d.g <iduartgomez@users.noreply.github.com>

Author: 3 people

crates/core/src/client_events/mod.rs

@@ -1155,34 +1155,37 @@ async fn process_open_request(
                             tracing::debug!(
                                 peer_id = %peer_id,
                                 key = %key,
-                                "Starting direct SUBSCRIBE operation (legacy mode)",
+                                "Starting direct SUBSCRIBE operation",
                             );
 
-                            // Legacy mode: direct operation without deduplication
-                            let op_id =
-                                crate::node::subscribe(op_manager.clone(), key, Some(client_id))
-                                    .await
-                                    .inspect_err(|err| {
-                                        tracing::error!("Subscribe error: {}", err);
-                                    })?;
-
-                            tracing::debug!(
-                                request_id = %request_id,
-                                transaction_id = %op_id,
-                                operation = "subscribe",
-                                "Request-Transaction correlation"
-                            );
+                            // Generate transaction, register first, then run op
+                            let tx = crate::message::Transaction::new::<
+                                crate::operations::subscribe::SubscribeMsg,
+                            >();
 
                             op_manager
                                 .ch_outbound
-                                .waiting_for_transaction_result(op_id, client_id, request_id)
+                                .waiting_for_transaction_result(tx, client_id, request_id)
                                 .await
                                 .inspect_err(|err| {
                                     tracing::error!(
                                         "Error waiting for transaction result: {}",
                                         err
                                     );
                                 })?;
+
+                            crate::node::subscribe_with_id(op_manager.clone(), key, None, Some(tx))
+                                .await
+                                .inspect_err(|err| {
+                                    tracing::error!("Subscribe error: {}", err);
+                                })?;
+
+                            tracing::debug!(
+                                request_id = %request_id,
+                                transaction_id = %tx,
+                                operation = "subscribe",
+                                "Request-Transaction correlation"
+                            );
                         }
                     }
                     _ => {