fix(tap-agent): stop global allocation typing; normalize per-sender with isLegacy + sender_type

  - Remove global allocation typing in Manager; keep a single raw watcher
  - Add per-sender normalization using Network Subgraph Allocation.isLegacy + escrow-derived sender_type
  - Pass typed Receiver<HashSet<AllocationId>> to SenderAccount; remove in-actor retyping

Author: neithanmo

crates/tap-agent/src/agent/sender_account.rs

@@ -11,7 +11,6 @@ use std::{
 use anyhow::Context;
 use bigdecimal::{num_bigint::ToBigInt, ToPrimitive};
 use futures::{stream, StreamExt};
-use indexer_allocation::Allocation;
 use indexer_monitor::{EscrowAccounts, SubgraphClient};
 use indexer_query::{
     closed_allocations::{self, ClosedAllocations},
@@ -275,9 +274,8 @@ pub struct SenderAccountArgs {
     pub sender_id: Address,
     /// Watcher that returns a list of escrow accounts for current indexer
     pub escrow_accounts: Receiver<EscrowAccounts>,
-    /// Raw watcher of open and recently closed allocations from Network Subgraph
-    /// We normalize per-sender to the correct variant (Legacy/Horizon)
-    pub indexer_allocations: Receiver<HashMap<Address, Allocation>>,
+    /// Watcher of normalized allocation IDs (Legacy/Horizon) for this sender type
+    pub indexer_allocations: Receiver<HashSet<AllocationId>>, 
     /// SubgraphClient of the escrow subgraph
     pub escrow_subgraph: &'static SubgraphClient,
     /// SubgraphClient of the network subgraph
@@ -830,34 +828,21 @@ impl Actor for SenderAccount {
             sender_type,
         }: Self::Arguments,
     ) -> Result<Self::State, ActorProcessingErr> {
-        // Normalize raw allocation addresses to the correct variant for this sender_type
+        // Pass-through normalized allocation IDs for this sender type
         let myself_clone = myself.clone();
-        let sender_type_for_log = sender_type; // copy for move into closure
-        watch_pipe(indexer_allocations, move |alloc_map| {
-            let raw_count = alloc_map.len();
-            // Extract addresses and normalize based on sender_type
-            let normalized: HashSet<AllocationId> = alloc_map
-                .keys()
-                .cloned()
-                .map(|addr| match sender_type_for_log {
-                    SenderType::Legacy => AllocationId::Legacy(AllocationIdCore::from(addr)),
-                    SenderType::Horizon => AllocationId::Horizon(CollectionId::from(addr)),
-                })
-                .collect();
-
+        watch_pipe(indexer_allocations, move |allocation_ids| {
+            let count = allocation_ids.len();
             tracing::info!(
                 sender = %sender_id,
-                ?sender_type_for_log,
-                raw_count,
-                normalized_count = normalized.len(),
-                "indexer_allocations update: normalizing allocations for sender_type",
+                sender_type = ?sender_type,
+                count,
+                "indexer_allocations update: received normalized allocations"
             );
-
-            // Forward normalized set to the actor to update allocations
             let myself = myself_clone.clone();
+            let allocation_ids = allocation_ids.clone();
             async move {
                 myself
-                    .cast(SenderAccountMessage::UpdateAllocationIds(normalized))
+                    .cast(SenderAccountMessage::UpdateAllocationIds(allocation_ids))
                     .unwrap_or_else(|e| {
                         tracing::error!(error=?e, "Error while updating allocation_ids");
                     });

crates/tap-agent/src/agent/sender_accounts_manager.rs

@@ -991,12 +991,57 @@ impl State {
             SenderType::Horizon => self.escrow_accounts_v2.clone(),
         };
 
+        // Build a normalized allocation watcher for this sender type using isLegacy flag
+        // from the Network Subgraph. Fallback: if the flag is missing, normalize by sender_type.
+        let indexer_allocations = {
+            let sender_type_for_log = sender_type;
+            map_watcher(self.indexer_allocations.clone(), move |alloc_map| {
+                let total = alloc_map.len();
+                let mut legacy_count = 0usize;
+                let mut horizon_count = 0usize;
+                let mut mismatched = 0usize;
+                let set: HashSet<AllocationId> = alloc_map
+                    .iter()
+                    .filter_map(|(addr, alloc)| {
+                        if alloc.is_legacy {
+                            legacy_count += 1;
+                            if matches!(sender_type_for_log, SenderType::Legacy) {
+                                Some(AllocationId::Legacy(AllocationIdCore::from(*addr)))
+                            } else {
+                                mismatched += 1;
+                                None
+                            }
+                        } else {
+                            horizon_count += 1;
+                            if matches!(sender_type_for_log, SenderType::Horizon) {
+                                Some(AllocationId::Horizon(CollectionId::from(*addr)))
+                            } else {
+                                mismatched += 1;
+                                None
+                            }
+                        }
+                    })
+                    .collect();
+
+                tracing::info!(
+                    ?sender_type_for_log,
+                    total,
+                    legacy = legacy_count,
+                    horizon = horizon_count,
+                    mismatched,
+                    normalized = set.len(),
+                    "Normalized indexer allocations using isLegacy"
+                );
+                set
+            })
+        };
+
         Ok(SenderAccountArgs {
             config: self.config,
             pgpool: self.pgpool.clone(),
             sender_id: *sender_id,
             escrow_accounts,
-            indexer_allocations: self.indexer_allocations.clone(),
+            indexer_allocations,
             escrow_subgraph: self.escrow_subgraph,
             network_subgraph: self.network_subgraph,
             domain_separator: self.domain_separator.clone(),

crates/tap-agent/src/test.rs

@@ -13,7 +13,6 @@ use actors::TestableActor;
 use anyhow::anyhow;
 use bigdecimal::num_bigint::BigInt;
 use indexer_config;
-use indexer_allocation::Allocation;
 use indexer_monitor::{DeploymentDetails, EscrowAccounts, SubgraphClient};
 use indexer_receipt::TapReceipt;
 use ractor::{concurrency::JoinHandle, Actor, ActorRef};
@@ -105,7 +104,7 @@ pub fn get_sender_account_config() -> &'static SenderAccountConfig {
 #[bon::builder]
 pub async fn create_sender_account(
     pgpool: PgPool,
-    #[builder(default = HashMap::new())] initial_allocations_raw: HashMap<Address, Allocation>,
+    #[builder(default = HashSet::new())] initial_allocation: HashSet<AllocationId>,
     #[builder(default = TRIGGER_VALUE)] rav_request_trigger_value: u128,
     #[builder(default = TRIGGER_VALUE)] max_amount_willing_to_lose_grt: u128,
     escrow_subgraph_endpoint: Option<&str>,
@@ -175,7 +174,7 @@ pub async fn create_sender_account(
         pgpool,
         sender_id: SENDER.1,
         escrow_accounts: escrow_accounts_rx,
-        indexer_allocations: watch::channel(initial_allocations_raw).1,
+        indexer_allocations: watch::channel(initial_allocation).1,
         escrow_subgraph,
         network_subgraph,
         domain_separator: TAP_EIP712_DOMAIN_SEPARATOR.clone(),