// PUT with subscribe=true (put.rs:562-566)
if subscribe && is_seeding_contract {
    // TODO: Make put operation atomic by linking it to the completion of this subscription request.
    // Currently we can't link one transaction to another transaction's result, which would be needed
    // to make this fully atomic. This should be addressed in a future refactoring.
    super::start_subscription_request(op_manager, key).await;  // ← Fire-and-forget
}

// Mark operation as finished
new_state = Some(PutState::Finished { key });  // ← Parent finishes immediately

// put.rs:562-564
// TODO: Make put operation atomic by linking it to the completion of this subscription request.
// Currently we can't link one transaction to another transaction's result, which would be needed
// to make this fully atomic. This should be addressed in a future refactoring.

// operations/mod.rs:310-318
async fn start_subscription_request(op_manager: &OpManager, key: ContractKey) {
    let sub_op = subscribe::start_op(key);  // ← New independent transaction
    if let Err(error) = subscribe::request_subscribe(op_manager, sub_op).await {
        tracing::warn!(%error, "Error subscribing to contract");  // ← Silent failure!
    }
}

// message.rs:35-38
#[derive(Serialize, Deserialize, PartialEq, Eq, Hash, Clone, Copy)]
pub struct Transaction {
    id: Ulid,  // ← No parent field
}

// node/op_state_manager.rs:54-63
#[derive(Default)]
struct Ops {
    connect: DashMap<Transaction, ConnectOp>,
    put: DashMap<Transaction, PutOp>,
    get: DashMap<Transaction, GetOp>,
    subscribe: DashMap<Transaction, SubscribeOp>,
    update: DashMap<Transaction, UpdateOp>,
    completed: DashSet<Transaction>,       // ← Flat completion tracking
    under_progress: DashSet<Transaction>,  // ← No hierarchical structure
}

// put.rs:60-62
pub(super) fn finalized(&self) -> bool {
    self.state.is_none() || matches!(self.state, Some(PutState::Finished { .. }))
    // ↑ Only checks PUT state, not whether subscription completed
}

// mod.rs:108-112
Ok(OperationResult {
    return_msg: None,
    state: Some(final_state),
}) if final_state.finalized() => {
    // operation finished_completely with result
    op_manager.completed(tx_id);  // ← Parent marked complete
    return Ok(Some(final_state));  // ← Client receives response
}

// node/mod.rs:413-421
if let Some(transaction) = tx {
    let host_result = op_res.to_host_result();  // ← Convert to client response
    let router_tx_clone = op_manager.result_router_tx.clone();

    tokio::spawn(async move {
        if let Err(e) = router_tx_clone.send((transaction, host_result)).await {
            // Send to client via result router
        }
    });
}

pub(super) fn finalized(&self) -> bool {
    self.state.is_none() || matches!(self.state, Some(PutState::Finished { .. }))
}

async fn start_subscription_request(op_manager: &OpManager, key: ContractKey) {
    let sub_op = subscribe::start_op(key);  // ← New independent transaction
    if let Err(error) = subscribe::request_subscribe(op_manager, sub_op).await {
        tracing::warn!(%error, "Error subscribing to contract");  // ← Logged and dropped
    }
}

// message.rs
pub struct Transaction {
    id: Ulid,
    parent: Option<Ulid>,  // ← NEW: Parent transaction ID
}

impl Transaction {
    pub fn new<T: TxType>() -> Self {
        // Existing implementation (no parent)
    }

    pub fn new_child_of(parent: &Transaction, tx_type: TransactionType) -> Self {
        let id = Ulid::new();
        let updated_id = Self::update(tx_type as u8, id);
        Self {
            id: updated_id,
            parent: Some(parent.id),  // ← Link to parent
        }
    }

    pub fn parent_id(&self) -> Option<&Ulid> {
        self.parent.as_ref()
    }
}

// node/op_state_manager.rs
use dashmap::DashMap;
use std::collections::HashSet;
use std::sync::Arc;

pub(crate) struct OpManager {
    // ... existing fields ...

    /// Maps parent transaction to set of child transactions
    sub_operations: Arc<DashMap<Transaction, HashSet<Transaction>>>,

    /// Operations that reached Finished state but have pending children
    pending_finalization: Arc<DashMap<Transaction, OpEnum>>,
}

impl OpManager {
    /// Register a sub-operation relationship
    pub fn register_sub_operation(&self, parent: Transaction, child: Transaction) {
        self.sub_operations
            .entry(parent)
            .or_insert_with(HashSet::new)
            .insert(child);

        tracing::debug!(
            parent_tx = %parent,
            child_tx = %child,
            "Registered sub-operation"
        );
    }

    /// Check if all sub-operations of a parent have completed
    pub fn all_sub_operations_completed(&self, parent: Transaction) -> bool {
        match self.sub_operations.get(&parent) {
            None => true,  // No sub-operations
            Some(children) => {
                // Check if all children are in completed set
                children.iter().all(|child| self.ops.completed.contains(child))
            }
        }
    }

    /// Mark a sub-operation as completed, check if parent can now finalize
    pub async fn sub_operation_completed(
        &self,
        child: Transaction,
    ) -> Result<Option<Transaction>, OpError> {
        // Mark child as completed (existing logic)
        self.completed(child);

        // Find parent transaction
        // Note: We need to iterate or maintain reverse mapping
        for entry in self.sub_operations.iter() {
            let (parent_tx, children) = entry.pair();
            if children.contains(&child) {
                // Check if ALL siblings are now complete
                if self.all_sub_operations_completed(*parent_tx) {
                    tracing::info!(
                        parent_tx = %parent_tx,
                        child_tx = %child,
                        "All sub-operations completed, parent can finalize"
                    );
                    return Ok(Some(*parent_tx));
                }
            }
        }
        Ok(None)
    }

    /// Handle sub-operation failure - propagate to parent
    pub async fn sub_operation_failed(
        &self,
        child: Transaction,
        error: OpError,
    ) -> Result<(), OpError> {
        tracing::error!(
            child_tx = %child,
            error = %error,
            "Sub-operation failed, propagating to parent"
        );

        // Find parent and fail it
        for entry in self.sub_operations.iter() {
            let (parent_tx, children) = entry.pair();
            if children.contains(&child) {
                // Mark parent as failed
                self.completed(*parent_tx);

                // Send error to client
                let error_result = Err(freenet_stdlib::client_api::ErrorKind::OperationError {
                    cause: format!("Sub-operation {} failed: {}", child, error).into(),
                }.into());

                self.result_router_tx
                    .send((*parent_tx, error_result))
                    .await
                    .map_err(|_| OpError::NotificationError)?;

                return Ok(());
            }
        }
        Ok(())
    }
}

/// Reverse index: child -> parent (for faster lookups)
parent_of: Arc<DashMap<Transaction, Transaction>>,

// operations/mod.rs
async fn handle_op_result<CB>(
    op_manager: &OpManager,
    network_bridge: &mut CB,
    result: Result<OperationResult, OpError>,
    tx_id: Transaction,
    sender: Option<PeerId>,
) -> Result<Option<OpEnum>, OpError>
where
    CB: NetworkBridge,
{
    match result {
        // ... error handling unchanged ...

        Ok(OperationResult {
            return_msg: None,
            state: Some(final_state),
        }) if final_state.finalized() => {
            // ↓ NEW: Check sub-operations before finalizing
            if op_manager.all_sub_operations_completed(tx_id) {
                // All sub-operations done, finalize immediately
                tracing::debug!(%tx_id, "operation completed with all sub-operations");
                op_manager.completed(tx_id);
                return Ok(Some(final_state));
            } else {
                // Parent reached finished state but children still pending
                tracing::debug!(
                    %tx_id,
                    "operation finished but awaiting sub-operation completion"
                );

                // Store in pending_finalization map
                op_manager.pending_finalization.insert(tx_id, final_state.clone());

                // Don't send response to client yet
                return Ok(None);
            }
        }

        // ... rest unchanged ...
    }
    Ok(None)
}

// In operation completion path
pub fn completed(&self, id: Transaction) {
    self.ring.live_tx_tracker.remove_finished_transaction(id);
    self.ops.completed.insert(id);

    // ↓ NEW: Check if this was a sub-operation
    if let Some(parent_tx) = self.parent_of.get(&id) {
        if self.all_sub_operations_completed(*parent_tx) {
            // All siblings completed, finalize parent now
            if let Some((_key, parent_op)) = self.pending_finalization.remove(&parent_tx) {
                tracing::info!(
                    parent_tx = %parent_tx,
                    "All sub-operations completed, finalizing parent"
                );

                // Send result to client
                let host_result = parent_op.to_host_result();
                let _ = self.result_router_tx.try_send((*parent_tx, host_result));
            }
        }
    }
}

// operations/mod.rs
async fn start_subscription_request(
    op_manager: &OpManager,
    parent_tx: Transaction,  // ← NEW: Parent transaction
    key: ContractKey,
) -> Result<Transaction, OpError> {  // ← NEW: Return child transaction
    // Create child transaction
    let child_tx = Transaction::new_child_of(&parent_tx, TransactionType::Subscribe);

    // Register with parent
    op_manager.register_sub_operation(parent_tx, child_tx);

    // Create subscription operation with child transaction
    let sub_op = subscribe::start_op_with_id(key, child_tx);

    // Execute subscription
    match subscribe::request_subscribe(op_manager, sub_op).await {
        Ok(_) => {
            tracing::debug!(
                parent_tx = %parent_tx,
                child_tx = %child_tx,
                "Subscription sub-operation succeeded"
            );
            Ok(child_tx)
        }
        Err(error) => {
            tracing::error!(
                parent_tx = %parent_tx,
                child_tx = %child_tx,
                error = %error,
                "Subscription sub-operation failed"
            );

            // ↓ NEW: Propagate failure to parent
            op_manager.sub_operation_failed(child_tx, error.clone()).await?;
            Err(error)
        }
    }
}

// put.rs:565
if subscribe && is_seeding_contract {
    // ↓ Changed: Pass parent transaction ID
    super::start_subscription_request(op_manager, id, key).await?;
    //                                              ↑ parent tx
}

#[tokio::test]
async fn test_transaction_parent_child_relationship() {
    let parent_tx = Transaction::new::<PutMsg>();
    let child_tx = Transaction::new_child_of(&parent_tx, TransactionType::Subscribe);

    assert_eq!(child_tx.parent_id(), Some(&parent_tx.id));
    assert_eq!(child_tx.transaction_type(), TransactionType::Subscribe);
}

#[tokio::test]
async fn test_sub_operation_registration() {
    let op_manager = create_test_op_manager().await;
    let parent_tx = Transaction::new::<PutMsg>();
    let child_tx = Transaction::new_child_of(&parent_tx, TransactionType::Subscribe);

    op_manager.register_sub_operation(parent_tx, child_tx);

    // Parent should not be complete yet
    assert!(!op_manager.all_sub_operations_completed(parent_tx));

    // Mark child complete
    op_manager.completed(child_tx);

    // Now parent should be ready to complete
    assert!(op_manager.all_sub_operations_completed(parent_tx));
}

#[tokio::test]
async fn test_put_subscribe_atomicity_success() {
    // Setup multi-node network
    let mut network = TestNetwork::new(3).await;

    // PUT with subscribe=true
    let result = network.nodes[0]
        .put_contract(contract, state, subscribe: true)
        .await;

    // Should succeed
    assert!(result.is_ok());

    // Verify subscription was established
    let subscriptions = network.nodes[0].get_subscriptions(&contract_key);
    assert!(subscriptions.contains(&contract_key));

    // Verify UPDATE can propagate
    let update_result = network.nodes[0]
        .update_contract(contract_key, new_state)
        .await;
    assert!(update_result.is_ok());
}

#[tokio::test]
async fn test_put_subscribe_atomicity_failure() {
    // Setup network where subscription will fail
    let mut network = TestNetwork::new_with_partition(3).await;
    network.partition_node(2);  // Isolate node 2

    // PUT with subscribe=true - should fail atomically
    let result = network.nodes[2]
        .put_contract(contract, state, subscribe: true)
        .await;

    // Should fail because subscription can't establish
    assert!(result.is_err());

    // Contract should NOT be stored (atomicity)
    let get_result = network.nodes[2].get_contract(&contract_key).await;
    assert!(get_result.is_err() || get_result.unwrap().is_none());
}

#[tokio::test]
async fn test_sub_operation_error_propagation() {
    let op_manager = create_test_op_manager().await;
    let parent_tx = Transaction::new::<PutMsg>();
    let child_tx = Transaction::new_child_of(&parent_tx, TransactionType::Subscribe);

    op_manager.register_sub_operation(parent_tx, child_tx);

    // Simulate child failure
    let error = OpError::RingError(RingError::EmptyRing);
    op_manager.sub_operation_failed(child_tx, error).await.unwrap();

    // Parent should be marked completed (with error)
    assert!(op_manager.ops.completed.contains(&parent_tx));

    // Error should be in result router
    let (tx, result) = timeout(Duration::from_millis(100),
                                op_manager.result_router_rx.recv())
        .await
        .unwrap()
        .unwrap();

    assert_eq!(tx, parent_tx);
    assert!(result.is_err());
}

#[tokio::test]
async fn test_nested_sub_operations() {
    // Parent -> Child1 -> Grandchild1
    //        -> Child2
    let op_manager = create_test_op_manager().await;
    let parent = Transaction::new::<PutMsg>();
    let child1 = Transaction::new_child_of(&parent, TransactionType::Subscribe);
    let child2 = Transaction::new_child_of(&parent, TransactionType::Get);
    let grandchild = Transaction::new_child_of(&child1, TransactionType::Get);

    op_manager.register_sub_operation(parent, child1);
    op_manager.register_sub_operation(parent, child2);
    op_manager.register_sub_operation(child1, grandchild);

    // Complete grandchild
    op_manager.completed(grandchild);
    assert!(op_manager.all_sub_operations_completed(child1));

    // Complete child1
    op_manager.completed(child1);
    assert!(!op_manager.all_sub_operations_completed(parent));  // child2 still pending

    // Complete child2
    op_manager.completed(child2);
    assert!(op_manager.all_sub_operations_completed(parent));  // Now parent can complete
}

#[tokio::test(flavor = "multi_thread")]
async fn test_river_chat_room_creation_atomic() {
    // Test the actual River use case from #1765
    let gateway = TestGateway::start().await;

    // Create room with PUT + subscribe
    let room_result = gateway.client()
        .create_room("test-room", subscribe: true)
        .await;

    assert!(room_result.is_ok(), "Room creation should succeed atomically");

    // Send message immediately (requires subscription to be working)
    let msg_result = gateway.client()
        .send_message("test-room", "Hello")
        .await;

    assert!(msg_result.is_ok(), "Message send should work immediately after room creation");
}

#[tokio::test(flavor = "multi_thread")]
async fn test_multi_hop_put_subscribe_atomicity() {
    // PUT routes through multiple peers
    let mut network = TestNetwork::new(5).await;

    // Client connected to node 0, optimal location at node 4
    let contract_key = ContractKey::from_params_and_code(...);
    let optimal_peer = network.nodes[4].peer_id();

    // PUT with subscribe from node 0
    let result = network.nodes[0]
        .put_contract(contract, state, subscribe: true)
        .await;

    assert!(result.is_ok());

    // Verify subscription established at all hops
    for i in 0..=4 {
        let subs = network.nodes[i].get_subscriptions(&contract_key);
        assert!(subs.contains(&contract_key),
                "Node {} should have subscription", i);
    }

    // Update from optimal location should reach all peers
    let update_sent = network.nodes[4]
        .broadcast_update(&contract_key, new_state)
        .await;

    assert_eq!(update_sent.len(), 4, "Should broadcast to 4 other peers");
}

#[tokio::test(flavor = "multi_thread")]
async fn test_get_proximity_caching_with_subscription() {
    let mut network = TestNetwork::new(3).await;

    // Node 0 has contract, node 1 gets it
    let contract_key = network.nodes[0].store_contract(contract, state).await;

    // Node 1 performs GET (should trigger proximity caching)
    let result = network.nodes[1].get_contract(&contract_key).await;
    assert!(result.is_ok());

    // Verify node 1 subscribed (for caching)
    let node1_subs = network.nodes[1].get_subscriptions(&contract_key);
    assert!(node1_subs.contains(&contract_key));

    // Update at node 0 should propagate to node 1
    network.nodes[0].update_contract(&contract_key, new_state).await;

    // Wait for propagation
    tokio::time::sleep(Duration::from_millis(100)).await;

    // Node 1 should have updated state
    let cached_state = network.nodes[1].get_cached_state(&contract_key).await;
    assert_eq!(cached_state, new_state);
}

enum OperationStatus {
    Pending,      // Primary operation complete, sub-operations running
    Confirmed,    // All sub-operations complete
    Failed,       // Sub-operation failed
}

parent_of: Arc<DashMap<Transaction, Transaction>>