feat(tap-agent): add sender account task logic and manager notifications

Author: suchapalaver

crates/tap-agent/src/actor_migrate.rs

@@ -99,6 +99,11 @@ impl<T> TaskHandle<T> {
             .map_err(|_| anyhow!("Task channel closed"))
     }
 
+    /// Send a message to the task (alias for cast)
+    pub async fn send(&self, msg: T) -> Result<()> {
+        self.cast(msg).await
+    }
+
     /// Stop the task
     pub fn stop(&self, _reason: Option<String>) {
         self.lifecycle.stop_task(self.task_id);
@@ -284,6 +289,7 @@ pub struct TaskContext {
 }
 
 /// Global task registry for named lookups
+#[derive(Clone)]
 pub struct TaskRegistry {
     registry: Arc<RwLock<HashMap<String, Box<dyn std::any::Any + Send + Sync>>>>,
 }
@@ -302,6 +308,7 @@ impl TaskRegistry {
     }
 
     /// Register a task handle
+    #[allow(dead_code)]
     pub async fn register<T>(&self, name: String, handle: TaskHandle<T>)
     where
         T: Send + Sync + 'static,
@@ -320,6 +327,14 @@ impl TaskRegistry {
             .get(name)
             .and_then(|any| any.downcast_ref::<TaskHandle<T>>().cloned())
     }
+
+    /// Get a task by name (alias for lookup)
+    pub async fn get_task<T>(&self, name: &str) -> Option<TaskHandle<T>>
+    where
+        T: Send + Sync + 'static,
+    {
+        self.lookup(name).await
+    }
 }
 
 /// Compatibility wrapper to make ractor ActorRef work with our abstraction

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

@@ -25,15 +25,19 @@ use tokio::sync::{mpsc, watch::Receiver};
 use super::{
     sender_account::{RavInformation, ReceiptFees, SenderAccountConfig, SenderAccountMessage},
     sender_accounts_manager::AllocationId,
-    sender_allocation_task::SenderAllocationTask,
     unaggregated_receipts::UnaggregatedReceipts,
 };
 use crate::{
     actor_migrate::{LifecycleManager, RestartPolicy, TaskHandle, TaskRegistry},
-    tap::context::{Horizon, Legacy},
     tracker::{SenderFeeTracker, SimpleFeeTracker},
 };
 
+#[cfg(any(test, feature = "test"))]
+use super::sender_allocation_task::SenderAllocationTask;
+
+#[cfg(any(test, feature = "test"))]
+use crate::tap::context::{Horizon, Legacy};
+
 type Balance = U256;
 type RavMap = HashMap<Address, u128>;
 
@@ -58,8 +62,10 @@ struct TaskState {
     /// Current sender balance
     sender_balance: U256,
     /// Registry for managing child tasks
+    #[allow(dead_code)]
     child_registry: TaskRegistry,
     /// Lifecycle manager for child tasks
+    #[allow(dead_code)]
     lifecycle: Arc<LifecycleManager>,
     /// Configuration
     #[allow(dead_code)]
@@ -339,10 +345,22 @@ impl SenderAccountTask {
 
         #[cfg(not(any(test, feature = "test")))]
         {
-            // In production, we'd need a proper way to create a self-reference
-            // For now, just log that this isn't implemented yet
+            // TODO: Implement production child task spawning
+            // This requires proper integration with the actual SenderAllocationTask spawn method
+            // that includes TAP manager, aggregator client, etc.
+            //
+            // For now, we'll skip this to maintain build compatibility
+            // The proper implementation would look like:
+            //
+            // 1. Create aggregator client for this sender
+            // 2. Set up TAP manager with proper configuration
+            // 3. Spawn SenderAllocationTask with full parameters
+            // 4. Set up proper parent-child communication channel
+
             tracing::warn!(
-                "Production sender allocation spawning not fully implemented yet - child task communication needs work"
+                sender = %state.sender,
+                allocation_id = ?allocation_id,
+                "Production sender allocation spawning not yet implemented - requires TAP manager integration"
             );
         }
 
@@ -412,11 +430,28 @@ impl SenderAccountTask {
 
     /// Handle invalid receipt fee updates
     async fn handle_update_invalid_receipt_fees(
-        _state: &mut TaskState,
-        _allocation_id: AllocationId,
-        _unaggregated_fees: UnaggregatedReceipts,
+        state: &mut TaskState,
+        allocation_id: AllocationId,
+        unaggregated_fees: UnaggregatedReceipts,
     ) -> Result<()> {
-        // Simplified implementation
+        let addr = match allocation_id {
+            AllocationId::Legacy(id) => id.into_inner(),
+            AllocationId::Horizon(id) => thegraph_core::AllocationId::from(id).into_inner(),
+        };
+
+        // Track invalid receipt fees in the tracker
+        state
+            .invalid_receipts_tracker
+            .update(addr, unaggregated_fees.value);
+
+        tracing::debug!(
+            sender = %state.sender,
+            allocation_id = ?allocation_id,
+            invalid_value = unaggregated_fees.value,
+            invalid_count = unaggregated_fees.counter,
+            "Updated invalid receipt fees for allocation"
+        );
+
         Ok(())
     }