feat(permission0): stream funnels (#146)

This patch adds a new operation for stream permissions that happens when
distributing the tokens. A "funnel" means that incoming streams for a
permission will be emitted as a new, unified stream ID under this
permission. The funneling mechanism allows localization and isolation of
tokens in complex streaming systems.

I had some trouble doing this as the initial solution was still not
clear to me. For some reason, I initially thought I'd have to alter the
entire accumulation logic, but it eventually came to me that I'd only
have to override when distributing. It took me a while to come to this
conclusion.

Closes CHAIN-136.

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

- New Features
- Introduced optional stream “funneling” to route distributions through
derived streams, improving separation and observability.
- Permission updates and delegation now support enabling/disabling
funnels; events now emit source and target stream IDs.
  - Enforced a per-permission funnel limit.

- Documentation
- Added a “Funneling Streams” section with diagrams and updated API
usage.

- Tests
- Added end-to-end funnel scenarios and updated existing cases to new
parameters.

- Chores
- Runtime upgrade with migration and storage version bump;
runtime/config updated to include funnel limits.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: saiintbrisson

.helix/languages.toml

@@ -2,4 +2,4 @@
 cargo.extraEnv = { SKIP_WASM_BUILD = "true" }
 cargo.features = ["runtime-benchmarks"]
 check.extraEnv = { SKIP_WASM_BUILD = "true" }
-check.overrideCommand = ["cargo", "check", "--message-format=json"]
+check.overrideCommand = ["cargo", "check", "--tests", "--all", "--message-format=json"]

docs/permission0.md

@@ -395,14 +395,16 @@ Batch creates multiple namespace permissions with the same properties but differ
 pub fn update_stream_permission(
     origin: OriginFor<T>,
     permission_id: PermissionId,
-    recipients: Option<Vec<(T::AccountId, u16)>>,
-    accumulating: Option<bool>,
-    recipient_manager: Option<T::AccountId>,
-    weight_setter: Option<T::AccountId>,
+    new_recipients: Option<BoundedBTreeMap<T::AccountId, u16, T::MaxRecipientsPerPermission>>,
+    new_streams: Option<BoundedBTreeMap<StreamId, Percent, T::MaxStreamsPerPermission>>,
+    new_distribution_control: Option<DistributionControl<T>>,
+    new_recipient_manager: Option<Option<T::AccountId>>,
+    new_weight_setter: Option<Option<T::AccountId>>,
+    funnel: Option<bool>,
 ) -> DispatchResult
 ```
 
-Allows delegator or authorized accounts to update stream permission properties, including recipients, accumulation state, and management roles.
+Allows delegator or authorized accounts to update stream permission properties, including recipients, stream allocations, distribution control, management roles, and funnel state. The `funnel` parameter controls whether distributions output the original stream IDs or a derived funnel stream ID to recipients.
 
 ### update_namespace_permission
 
@@ -528,7 +530,76 @@ When streams are redelegated through the permission system, their IDs are preser
 
 This stream-based model allows for much more granular control over stream delegation, enabling agents to specify different delegation percentages for different types of streams they receive.
 
-## Integration with Stream Distribution
+## Funneling Streams
+
+While the stream-based model preserves lineage across delegations, complex multi-level swarms reveal a limitation: when multiple permissions route different portions of the same stream to an agent, all incoming tokens accumulate into a single bucket at `(Agent, StreamId)`. This mixes otherwise distinct delegation paths into one mixed balance, obscuring provenance and preventing agents from treating tokens differently based on their intended purpose or origin.
+
+Consider an agent that performs multiple specialized functions for the same swarm. A research collective with a root stream of 1000 TORS creates two permissions: one allocating 30% for compute services, another allocating 20% for data curation. Both permissions reference the same root stream, so the agent receives 500 TORS total, but cannot distinguish which 300 TORS arrived for compute work versus which 200 TORS came for curation. The agent cannot route compute emissions to GPU infrastructure while directing curation payments to data validators. It cannot apply function-specific fee structures, cannot generate separate invoices for each service, and cannot prove to sub-delegators how much funding flows through each functional role.
+
+This problem compounds when the agent needs to create child permissions. If it wants to forward 50% of compute earnings to a GPU cluster operator and 30% of curation earnings to a dataset validator, the mixed bucket prevents such functional isolation. The semantic separation needed for proper accounting, specialized routing, and role-based policy enforcement is lost.
+
+```mermaid
+graph TB
+    ROOT[Research Collective<br/>RootStream: 1000 TORS]
+
+    ROOT -->|Permission 1: 30%<br/>For Compute<br/>StreamID: 0xABC...| A[Multi-Function Agent]
+    ROOT -->|Permission 2: 20%<br/>For Curation<br/>StreamID: 0xABC...| A
+
+    A -->|Total: 500 TORS<br/>Mixed bucket<br/>Cannot distinguish| MIXED[Indistinguishable<br/>300 TORS compute + 200 TORS curation<br/>Origin lost]
+```
+
+Stream funnels solve this by transforming input streams into new derived Stream IDs that represent specific delegation paths. When a delegator creates a permission with `enable_funnel: true`, the system generates a unique derived Stream ID by hashing the permission ID. This derived stream becomes the output identifier that recipients see and accumulate, effectively creating a new stream identity that carries the semantic context of that delegation path.
+
+With funnels, the research collective creates two separate funnel permissions: one for compute services outputting `derived_stream_compute`, another for curation outputting `derived_stream_curation`. The agent now receives two isolated buckets, each carrying its own functional context. It can create child permissions on `derived_stream_compute` to route compute tokens to GPU providers, and separate permissions on `derived_stream_curation` for data validators. Each sub-delegator knows precisely which functional allocation their tokens originated from, without needing to query complex delegation graphs or reconstruct paths off-chain.
+
+```mermaid
+graph TB
+    ROOT[Research Collective<br/>RootStream: 1000 TORS]
+
+    ROOT -->|Permission 1 with Funnel<br/>Input: RootStream 30%| FC[Funnel: Compute<br/>Outputs: derived_compute]
+    ROOT -->|Permission 2 with Funnel<br/>Input: RootStream 20%| FD[Funnel: Curation<br/>Outputs: derived_curation]
+
+    FC -->|300 TORS<br/>derived_compute| A[Multi-Function Agent]
+    FD -->|200 TORS<br/>derived_curation| A
+
+    A -->|derived_compute<br/>50% sub-delegation| GPU[GPU Clusters<br/>Model training]
+    A -->|derived_curation<br/>30% sub-delegation| VAL[Data Validators<br/>Dataset verification]
+```
+
+Funnels preserve composability throughout the delegation chain. An agent receiving `derived_stream_compute` can create its own permissions on that stream, including another funnel that further specializes the path (for example, separating training workloads from inference services). Revocation works cleanly: revoking a funnel permission stops emissions along that path, but child permissions remain structurally valid and simply stop receiving funds. No cascade deletions occur, no coordination is required between levels.
+
+The funnel structure is minimal, containing only what's necessary for deterministic ID generation:
+
+```rust
+pub struct FunnelStream<T: Config> {
+    pub derived_stream: StreamId,
+    pub nonce: u64,
+    pub created_at: BlockNumberFor<T>,
+}
+```
+
+The derived Stream ID is computed as `hash(permission_id, nonce)` where the nonce is always 1 for the first funnel on a permission. This deterministic generation means any observer can reconstruct the derived stream ID from the permission ID without additional storage lookups. The `created_at` block number provides an audit trail for when the funnel was established.
+
+A stream permission can enable or disable funneling through the `update_stream_permission` extrinsic by passing `funnel: Some(true)` or `funnel: Some(false)`. Accumulation always occurs at the original input stream buckets regardless of funnel state, the funnel only transforms which stream ID recipients see during distribution. Enabling a funnel on a permission that already has accumulated amounts means the next distribution will pass the derived stream ID to recipients instead of the original. Disabling a funnel means recipients will receive the original stream IDs on subsequent distributions.
+
+During distribution, the system reads accumulated amounts from original input stream buckets (where accumulation occurs), but passes the derived Stream ID to recipients through `do_distribute_to_targets`. Recipients accumulate tokens under `(Agent, derived_stream_id, permission_id)` buckets, achieving full separation from other permissions.
+
+Events capture both source and target streams for complete observability:
+
+```rust
+Event::StreamDistribution {
+    permission_id: H256,
+    source_stream: Some(0xROOT...),          // Original input stream
+    target_stream: Some(0xDERIVED_COMPUTE...), // Funnel's derived output
+    recipient: AgentAccount,
+    amount: 300_TORS,
+    reason: DistributionReason,
+}
+```
+
+This dual-stream event structure allows indexers to reconstruct the full flow without maintaining complex state. Accumulation events show original streams, distribution events show both source and target, enabling path tracing from root streams through arbitrary delegation depths. Off-chain systems can identify funnel operations by comparing source and target streams, when they differ, a funnel transformation occurred.
+
+## Hooking into emission rewards
 
 Permission0 integrates with the `Emission0` pallet by intercepting the stream distribution process. When the linear rewards mechanism distributes tokens, the `do_accumulate_streams` function is called to divert portions according to active permissions.
 

pallets/permission0/api/src/lib.rs

@@ -26,6 +26,17 @@ pub fn generate_root_stream_id<AccountId: Encode>(agent_id: &AccountId) -> Strea
     blake2_256(&data).into()
 }
 
+/// Static identifier for funneled streams
+pub const FUNNEL_STREAM_PREFIX: &[u8] = b"torus:permission:funnel-v1";
+
+/// Generates a funneled stream ID.
+pub fn generate_funnel_stream_id(permission_id: &PermissionId, nonce: u32) -> StreamId {
+    let mut data = FUNNEL_STREAM_PREFIX.to_vec();
+    data.extend(permission_id.encode());
+    data.extend(nonce.encode());
+    blake2_256(&data).into()
+}
+
 /// Defines what portion of streams the permission applies to
 #[derive(Encode, Decode, TypeInfo, Clone, PartialEq, Eq, Debug)]
 pub enum StreamAllocation<Balance> {
@@ -113,6 +124,7 @@ pub trait Permission0StreamApi<AccountId, Origin, BlockNumber, Balance, Negative
         enforcement: EnforcementAuthority<AccountId>,
         recipient_manager: Option<AccountId>,
         weight_setter: Option<AccountId>,
+        enable_funnel: bool,
     ) -> Result<PermissionId, DispatchError>;
 
     /// Accumulate streams for an agent with permissions

pallets/permission0/src/ext/stream_impl.rs

@@ -43,6 +43,7 @@ impl<T: Config>
         enforcement: ApiEnforcementAuthority<T::AccountId>,
         recipient_manager: Option<T::AccountId>,
         weight_setter: Option<T::AccountId>,
+        enable_funnel: bool,
     ) -> Result<PermissionId, DispatchError> {
         let internal_allocation = match allocation {
             ApiStreamAllocation::Streams(streams) => StreamAllocation::Streams(
@@ -81,6 +82,7 @@ impl<T: Config>
             enforcement,
             recipient_manager,
             weight_setter,
+            enable_funnel,
         )
     }
 
@@ -121,6 +123,7 @@ pub(crate) fn delegate_stream_permission_impl<T: Config>(
     enforcement: EnforcementAuthority<T>,
     recipient_manager: Option<T::AccountId>,
     weight_setter: Option<T::AccountId>,
+    enable_funnel: bool,
 ) -> Result<PermissionId, DispatchError> {
     use polkadot_sdk::frame_support::ensure;
 
@@ -155,17 +158,26 @@ pub(crate) fn delegate_stream_permission_impl<T: Config>(
 
     let recipients_ids: Vec<_> = recipients.keys().cloned().collect();
 
-    let scope = PermissionScope::Stream(StreamScope {
+    let mut scope = PermissionScope::Stream(StreamScope {
         recipients,
         allocation: allocation.clone(),
         distribution,
-        accumulating: true, // Start with accumulation enabled by default
+        accumulating: true,
         recipient_managers: validate_stream_managers::<T>(&delegator, recipient_manager)?,
         weight_setters: validate_stream_managers::<T>(&delegator, weight_setter)?,
+        funnels: Default::default(),
     });
 
     let permission_id = generate_permission_id::<T>(&delegator, &scope)?;
 
+    if enable_funnel {
+        let PermissionScope::Stream(stream) = &mut scope else {
+            return Err(Error::<T>::UnsupportedPermissionType.into());
+        };
+
+        stream.enable_funnel(permission_id)?;
+    }
+
     let contract =
         PermissionContract::<T>::new(delegator.clone(), scope, duration, revocation, enforcement);
 
@@ -316,6 +328,7 @@ pub(crate) fn update_stream_permission<T: Config>(
     new_distribution_control: Option<DistributionControl<T>>,
     new_recipient_manager: Option<Option<T::AccountId>>,
     new_weight_setter: Option<Option<T::AccountId>>,
+    funnel: Option<bool>,
 ) -> DispatchResult {
     let caller = ensure_signed(origin)?;
 
@@ -421,6 +434,20 @@ pub(crate) fn update_stream_permission<T: Config>(
             validate_stream_managers::<T>(&permission.delegator, new_weight_setter)?;
     }
 
+    if let Some(funnel) = funnel {
+        ensure!(allowed_delegator, Error::<T>::NotAuthorizedToEdit);
+        ensure!(
+            matches!(scope.allocation, StreamAllocation::Streams(_)),
+            Error::<T>::UnsupportedPermissionType
+        );
+
+        if funnel {
+            let _ = scope.enable_funnel(permission_id);
+        } else {
+            let _ = scope.disable_funnel(permission_id);
+        }
+    }
+
     permission.scope = PermissionScope::Stream(scope);
     permission.last_update = frame_system::Pallet::<T>::block_number();
     Permissions::<T>::set(permission_id, Some(permission));

pallets/permission0/src/lib.rs

@@ -39,7 +39,7 @@ pub mod pallet {
 
     use super::*;
 
-    const STORAGE_VERSION: StorageVersion = StorageVersion::new(7);
+    const STORAGE_VERSION: StorageVersion = StorageVersion::new(8);
 
     /// Configure the pallet by specifying the parameters and types on which it depends.
     #[pallet::config]
@@ -220,7 +220,11 @@ pub mod pallet {
         /// An stream distribution happened
         StreamDistribution {
             permission_id: PermissionId,
-            stream_id: Option<StreamId>,
+            /// The source stream ID from which the tokens were derived.
+            source_stream: Option<StreamId>,
+            /// The actual final stream ID to which tokens were accumulated,
+            /// this is the funnel stream ID if enabled.
+            target_stream: Option<StreamId>,
             recipient: T::AccountId,
             amount: BalanceOf<T>,
             reason: permission::stream::DistributionReason,
@@ -231,6 +235,16 @@ pub mod pallet {
             stream_id: StreamId,
             amount: BalanceOf<T>,
         },
+        /// Funnel enabled for permission
+        FunnelEnabled {
+            permission_id: PermissionId,
+            derived_stream_id: StreamId,
+        },
+        /// Funnel disabled for permission
+        FunnelDisabled {
+            permission_id: PermissionId,
+            derived_stream_id: StreamId,
+        },
     }
 
     #[pallet::error]
@@ -329,6 +343,8 @@ pub mod pallet {
         TooManyCuratorPermissions,
         /// Namespace delegation depth exceeded the maximum allowed limit.
         DelegationDepthExceeded,
+        /// Current funnel system is limited to one derived stream per permission.
+        TooManyStreamFunnels,
     }
 
     #[pallet::hooks]
@@ -353,6 +369,7 @@ pub mod pallet {
             enforcement: EnforcementAuthority<T>,
             recipient_manager: Option<T::AccountId>,
             weight_setter: Option<T::AccountId>,
+            enable_funnel: bool,
         ) -> DispatchResult {
             let delegator = ensure_signed(origin)?;
 
@@ -366,6 +383,7 @@ pub mod pallet {
                 enforcement,
                 recipient_manager,
                 weight_setter,
+                enable_funnel,
             )?;
 
             Ok(())
@@ -560,6 +578,7 @@ pub mod pallet {
             new_distribution_control: Option<DistributionControl<T>>,
             new_recipient_manager: Option<Option<T::AccountId>>,
             new_weight_setter: Option<Option<T::AccountId>>,
+            funnel: Option<bool>,
         ) -> DispatchResult {
             ext::stream_impl::update_stream_permission(
                 origin,
@@ -569,6 +588,7 @@ pub mod pallet {
                 new_distribution_control,
                 new_recipient_manager,
                 new_weight_setter,
+                funnel,
             )?;
 
             Ok(())

pallets/permission0/src/migrations.rs

@@ -1 +1,103 @@
+use polkadot_sdk::frame_support::{
+    migrations::VersionedMigration, pallet_prelude::*, traits::UncheckedOnRuntimeUpgrade,
+};
 
+use crate::{Config, PermissionId, Permissions};
+
+pub mod v8 {
+
+    use super::*;
+
+    pub type MigrationToV8<T, W> = VersionedMigration<7, 8, MigrateToV8<T>, crate::Pallet<T>, W>;
+
+    pub struct MigrateToV8<T>(core::marker::PhantomData<T>);
+
+    mod old {
+        use codec::{Decode, Encode};
+        use polkadot_sdk::{
+            polkadot_sdk_frame::prelude::BlockNumberFor,
+            sp_runtime::{BoundedBTreeMap, BoundedBTreeSet},
+        };
+
+        use super::*;
+        use crate::permission::*;
+
+        #[derive(Encode, Decode)]
+        pub struct StreamScope<T: Config> {
+            pub recipients: BoundedBTreeMap<T::AccountId, u16, T::MaxRecipientsPerPermission>,
+            pub allocation: StreamAllocation<T>,
+            pub distribution: DistributionControl<T>,
+            pub accumulating: bool,
+            pub recipient_managers: BoundedBTreeSet<T::AccountId, T::MaxControllersPerPermission>,
+            pub weight_setters: BoundedBTreeSet<T::AccountId, T::MaxControllersPerPermission>,
+        }
+
+        #[derive(Encode, Decode)]
+        pub enum PermissionScope<T: Config> {
+            Stream(StreamScope<T>),
+            Curator(CuratorScope<T>),
+            Namespace(NamespaceScope<T>),
+            Wallet(wallet::WalletScope<T>),
+        }
+
+        #[derive(Encode, Decode)]
+        pub struct PermissionContract<T: Config> {
+            pub delegator: T::AccountId,
+            pub scope: PermissionScope<T>,
+            pub duration: PermissionDuration<T>,
+            pub revocation: RevocationTerms<T>,
+            pub enforcement: EnforcementAuthority<T>,
+            pub last_update: BlockNumberFor<T>,
+            pub last_execution: Option<BlockNumberFor<T>>,
+            pub execution_count: u32,
+            pub created_at: BlockNumberFor<T>,
+        }
+    }
+
+    impl<T: Config> UncheckedOnRuntimeUpgrade for MigrateToV8<T> {
+        fn on_runtime_upgrade() -> Weight {
+            let mut weight = Weight::zero();
+            let mut translated = 0u64;
+
+            Permissions::<T>::translate(|_key: PermissionId, old: old::PermissionContract<T>| {
+                weight = weight.saturating_add(T::DbWeight::get().reads_writes(1, 1));
+                translated = translated.saturating_add(1);
+
+                let new_scope = match old.scope {
+                    old::PermissionScope::Stream(old_stream) => {
+                        crate::PermissionScope::Stream(crate::StreamScope {
+                            recipients: old_stream.recipients,
+                            allocation: old_stream.allocation,
+                            distribution: old_stream.distribution,
+                            accumulating: old_stream.accumulating,
+                            recipient_managers: old_stream.recipient_managers,
+                            weight_setters: old_stream.weight_setters,
+                            funnels: BoundedVec::default(),
+                        })
+                    }
+                    old::PermissionScope::Curator(curator) => {
+                        crate::PermissionScope::Curator(curator)
+                    }
+                    old::PermissionScope::Namespace(namespace) => {
+                        crate::PermissionScope::Namespace(namespace)
+                    }
+                    old::PermissionScope::Wallet(wallet) => crate::PermissionScope::Wallet(wallet),
+                };
+
+                Some(crate::PermissionContract {
+                    delegator: old.delegator,
+                    scope: new_scope,
+                    duration: old.duration,
+                    revocation: old.revocation,
+                    enforcement: old.enforcement,
+                    last_update: old.last_update,
+                    last_execution: old.last_execution,
+                    execution_count: old.execution_count,
+                    created_at: old.created_at,
+                })
+            });
+
+            weight
+        }
+    }
+}