diff --git a/docs/action-queue.md b/docs/action-queue.md
index 5a882c4d7..75687cc44 100644
--- a/docs/action-queue.md
+++ b/docs/action-queue.md
@@ -8,8 +8,8 @@ The action execution worker will only grab items from the action queue to execut
 
 ## Allocation management modes:
 - `auto`: The indexer-agent will act similarly to the legacy paradigm. When it identifies allocation actions it will add them to the queue with ActionStatus = `approved`; the execution worker process will pick up the approved actions within 30 seconds and execute them.
-- `manual`: The indexer-agent will not add any items to the action queue in this mode. It will spin up an indexer-management server which can be interacted with manually or integrated with 3rd party tools to add actions to the action queue and execute them.
-- `oversight`: The indexer-agent will add run its reconciliation loop to make allocation decisions and when actions are identified it will queue them. These actions will then require approval before they can be executed.
+- `manual`: The indexer-agent will not add any items to the action queue in this mode. It will spin up an indexer-management server which can be interacted with manually or integrated with 3rd party tools to add actions to the action queue and execute them. An exception to this is indexing agreements (DIPs), for which actions will be queued and executed even in this mode.
+- `oversight`: The indexer-agent will add run its reconciliation loop to make allocation decisions and when actions are identified it will queue them. These actions will then require approval before they can be executed. An exception to this is indexing agreements (DIPs), for which actions will be queued as approved and executed even in this mode.
 
 ## Actions CLI
 The indexer-cli provides an `actions` module for manually working with the action queue. It uses the #Graphql API hosted by the indexer management server to interact with the actions queue.
diff --git a/packages/indexer-agent/src/__tests__/indexer.ts b/packages/indexer-agent/src/__tests__/indexer.ts
index 419c90556..566412a0e 100644
--- a/packages/indexer-agent/src/__tests__/indexer.ts
+++ b/packages/indexer-agent/src/__tests__/indexer.ts
@@ -146,6 +146,7 @@ const setup = async () => {
   const network = await Network.create(
     logger,
     networkSpecification,
+    models,
     queryFeeModels,
     graphNode,
     metrics,
diff --git a/packages/indexer-agent/src/agent.ts b/packages/indexer-agent/src/agent.ts
index 166756623..fb55bae03 100644
--- a/packages/indexer-agent/src/agent.ts
+++ b/packages/indexer-agent/src/agent.ts
@@ -220,6 +220,16 @@ export class Agent {
       sequentialTimerMap(
         { logger, milliseconds: requestIntervalSmall },
         async () => {
+          if (network.specification.indexerOptions.enableDips) {
+            // There should be a DipsManager in the operator
+            if (!operator.dipsManager) {
+              throw new Error('DipsManager is not available')
+            }
+            logger.trace('Ensuring indexing rules for DIPS', {
+              protocolNetwork: network.specification.networkIdentifier,
+            })
+            await operator.dipsManager.ensureAgreementRules()
+          }
           logger.trace('Fetching indexing rules', {
             protocolNetwork: network.specification.networkIdentifier,
           })
@@ -252,14 +262,15 @@ export class Agent {
         },
       )
 
-    // Skip fetching active deployments if the deployment management mode is manual and POI tracking is disabled
+    // Skip fetching active deployments if the deployment management mode is manual, DIPs is disabled, and POI tracking is disabled
     const activeDeployments: Eventual<SubgraphDeploymentID[]> =
       sequentialTimerMap(
         { logger, milliseconds: requestIntervalLarge },
         async () => {
           if (
             this.deploymentManagement === DeploymentManagementMode.AUTO ||
-            network.networkMonitor.poiDisputeMonitoringEnabled()
+            network.networkMonitor.poiDisputeMonitoringEnabled() ||
+            network.specification.indexerOptions.enableDips
           ) {
             logger.trace('Fetching active deployments')
             const assignments =
@@ -487,9 +498,40 @@ export class Agent {
             }
             break
           case DeploymentManagementMode.MANUAL:
-            this.logger.debug(
-              `Skipping subgraph deployment reconciliation since DeploymentManagementMode = 'manual'`,
-            )
+            if (network.specification.indexerOptions.enableDips) {
+              // Reconcile DIPs deployments anyways
+              this.logger.warn(
+                `Deployment management is manual, but DIPs is enabled. Reconciling DIPs deployments anyways.`,
+              )
+              if (!operator.dipsManager) {
+                throw new Error('DipsManager is not available')
+              }
+              const dipsDeployments =
+                await operator.dipsManager.getActiveDipsDeployments()
+              const newTargetDeployments = new Set([
+                ...activeDeployments,
+                ...dipsDeployments,
+              ])
+              try {
+                await this.reconcileDeployments(
+                  activeDeployments,
+                  Array.from(newTargetDeployments),
+                  eligibleAllocations,
+                )
+              } catch (err) {
+                logger.warn(
+                  `Exited early while reconciling deployments. Skipped reconciling actions.`,
+                  {
+                    err: indexerError(IndexerErrorCode.IE005, err),
+                  },
+                )
+                return
+              }
+            } else {
+              this.logger.debug(
+                `Skipping subgraph deployment reconciliation since DeploymentManagementMode = 'manual'`,
+              )
+            }
             break
           default:
             throw new Error(
@@ -810,6 +852,7 @@ export class Agent {
     maxAllocationEpochs: number,
     network: Network,
     operator: Operator,
+    forceAction: boolean = false,
   ): Promise<void> {
     const logger = this.logger.child({
       deployment: deploymentAllocationDecision.deployment.ipfsHash,
@@ -831,6 +874,7 @@ export class Agent {
           logger,
           deploymentAllocationDecision,
           activeDeploymentAllocations,
+          forceAction,
         )
       case true: {
         // If no active allocations and subgraph health passes safety check, create one
@@ -867,6 +911,7 @@ export class Agent {
               logger,
               deploymentAllocationDecision,
               mostRecentlyClosedAllocation,
+              forceAction,
             )
           }
         } else if (activeDeploymentAllocations.length > 0) {
@@ -875,6 +920,7 @@ export class Agent {
               logger,
               deploymentAllocationDecision,
               activeDeploymentAllocations,
+              forceAction,
             )
           } else {
             // Refresh any expiring allocations
@@ -891,6 +937,7 @@ export class Agent {
                 logger,
                 deploymentAllocationDecision,
                 expiringAllocations,
+                forceAction,
               )
             }
           }
@@ -910,21 +957,37 @@ export class Agent {
     // --------------------------------------------------------------------------------
     const { network, operator } = this.networkAndOperator
     let validatedAllocationDecisions = [...allocationDecisions]
+    let dipsDeployments: SubgraphDeploymentID[] = []
+    if (network.specification.indexerOptions.enableDips) {
+      if (!operator.dipsManager) {
+        throw new Error('DipsManager is not available')
+      }
+      dipsDeployments = await operator.dipsManager.getActiveDipsDeployments()
+    }
 
     if (
       network.specification.indexerOptions.allocationManagementMode ===
       AllocationManagementMode.MANUAL
     ) {
-      this.logger.debug(
-        `Skipping allocation reconciliation since AllocationManagementMode = 'manual'`,
-        {
-          protocolNetwork: network.specification.networkIdentifier,
-          targetDeployments: allocationDecisions
-            .filter(decision => decision.toAllocate)
-            .map(decision => decision.deployment.ipfsHash),
-        },
-      )
-      validatedAllocationDecisions = [] as AllocationDecision[]
+      if (network.specification.indexerOptions.enableDips) {
+        this.logger.warn(
+          `Allocation management is manual, but DIPs is enabled. Reconciling DIPs allocations anyways.`,
+        )
+        validatedAllocationDecisions = validatedAllocationDecisions.filter(
+          decision => dipsDeployments.includes(decision.deployment),
+        )
+      } else {
+        this.logger.trace(
+          `Skipping allocation reconciliation since AllocationManagementMode = 'manual'`,
+          {
+            protocolNetwork: network.specification.networkIdentifier,
+            targetDeployments: allocationDecisions
+              .filter(decision => decision.toAllocate)
+              .map(decision => decision.deployment.ipfsHash),
+          },
+        )
+        validatedAllocationDecisions = [] as AllocationDecision[]
+      }
     } else {
       const networkSubgraphDeployment = network.networkSubgraph.deployment
       if (
@@ -985,6 +1048,7 @@ export class Agent {
         maxAllocationEpochs,
         network,
         operator,
+        dipsDeployments.includes(decision.deployment), // Force actions if this is a DIPs deployment
       ),
     )
     return
diff --git a/packages/indexer-agent/src/commands/start.ts b/packages/indexer-agent/src/commands/start.ts
index 8ffd97fbf..558d16b36 100644
--- a/packages/indexer-agent/src/commands/start.ts
+++ b/packages/indexer-agent/src/commands/start.ts
@@ -308,6 +308,26 @@ export const start = {
         default: 1,
         group: 'Indexer Infrastructure',
       })
+      .option('enable-dips', {
+        description: 'Whether to enable Indexing Fees (DIPs)',
+        type: 'boolean',
+        default: false,
+        group: 'Indexing Fees ("DIPs")',
+      })
+      .option('dipper-endpoint', {
+        description: 'Gateway endpoint for DIPs receipts',
+        type: 'string',
+        array: false,
+        required: false,
+        group: 'Indexing Fees ("DIPs")',
+      })
+      .option('dips-allocation-amount', {
+        description: 'Amount of GRT to allocate for DIPs',
+        type: 'number',
+        default: 1,
+        required: false,
+        group: 'Indexing Fees ("DIPs")',
+      })
       .check(argv => {
         if (
           !argv['network-subgraph-endpoint'] &&
@@ -335,6 +355,9 @@ export const start = {
         ) {
           return 'Invalid --rebate-claim-max-batch-size provided. Must be > 0 and an integer.'
         }
+        if (argv['enable-dips'] && !argv['dipper-endpoint']) {
+          return 'Invalid --dipper-endpoint provided. Must be provided when --enable-dips is true.'
+        }
         return true
       })
   },
@@ -370,6 +393,10 @@ export async function createNetworkSpecification(
     allocateOnNetworkSubgraph: argv.allocateOnNetworkSubgraph,
     register: argv.register,
     finalityTime: argv.chainFinalizeTime,
+    enableDips: argv.enableDips,
+    dipperEndpoint: argv.dipperEndpoint,
+    dipsAllocationAmount: argv.dipsAllocationAmount,
+    dipsEpochsMargin: argv.dipsEpochsMargin,
   }
 
   const transactionMonitoring = {
@@ -587,6 +614,7 @@ export async function run(
   const network = await Network.create(
     logger,
     networkSpecification,
+    managementModels,
     queryFeeModels,
     graphNode,
     metrics,
diff --git a/packages/indexer-cli/src/__tests__/util.ts b/packages/indexer-cli/src/__tests__/util.ts
index 502bc9689..e9957ff48 100644
--- a/packages/indexer-cli/src/__tests__/util.ts
+++ b/packages/indexer-cli/src/__tests__/util.ts
@@ -87,6 +87,7 @@ export const setup = async () => {
   const network = await Network.create(
     logger,
     testNetworkSpecification,
+    models,
     queryFeeModels,
     graphNode,
     metrics,
diff --git a/packages/indexer-common/package.json b/packages/indexer-common/package.json
index 1a4141d36..e71aeaa80 100644
--- a/packages/indexer-common/package.json
+++ b/packages/indexer-common/package.json
@@ -22,7 +22,10 @@
     "clean": "rm -rf ./node_modules ./dist ./tsconfig.tsbuildinfo"
   },
   "dependencies": {
+    "@bufbuild/protobuf": "2.2.3",
     "@graphprotocol/common-ts": "2.0.11",
+    "@graphprotocol/dips-proto": "0.2.2",
+    "@grpc/grpc-js": "^1.12.6",
     "@pinax/graph-networks-registry": "0.6.7",
     "@semiotic-labs/tap-contracts-bindings": "^1.2.1",
     "@thi.ng/heaps": "1.2.38",
diff --git a/packages/indexer-common/src/allocations/__tests__/tap.test.ts b/packages/indexer-common/src/allocations/__tests__/tap.test.ts
index b096a9220..43d746190 100644
--- a/packages/indexer-common/src/allocations/__tests__/tap.test.ts
+++ b/packages/indexer-common/src/allocations/__tests__/tap.test.ts
@@ -7,6 +7,7 @@ import {
   TapSubgraphResponse,
   TapCollector,
   Allocation,
+  defineIndexerManagementModels,
 } from '@graphprotocol/indexer-common'
 import {
   Address,
@@ -43,6 +44,7 @@ const setup = async () => {
   // Clearing the registry prevents duplicate metric registration in the default registry.
   metrics.registry.clear()
   sequelize = await connectDatabase(__DATABASE__)
+  const models = defineIndexerManagementModels(sequelize)
   queryFeeModels = defineQueryFeeModels(sequelize)
   sequelize = await sequelize.sync({ force: true })
 
@@ -56,6 +58,7 @@ const setup = async () => {
   const network = await Network.create(
     logger,
     testNetworkSpecification,
+    models,
     queryFeeModels,
     graphNode,
     metrics,
diff --git a/packages/indexer-common/src/allocations/__tests__/validate-queries.test.ts b/packages/indexer-common/src/allocations/__tests__/validate-queries.test.ts
index dbfec4b11..487fe70c4 100644
--- a/packages/indexer-common/src/allocations/__tests__/validate-queries.test.ts
+++ b/packages/indexer-common/src/allocations/__tests__/validate-queries.test.ts
@@ -1,4 +1,5 @@
 import {
+  defineIndexerManagementModels,
   defineQueryFeeModels,
   GraphNode,
   Network,
@@ -36,6 +37,7 @@ const setup = async () => {
   // Clearing the registry prevents duplicate metric registration in the default registry.
   metrics.registry.clear()
   sequelize = await connectDatabase(__DATABASE__)
+  const models = defineIndexerManagementModels(sequelize)
   queryFeeModels = defineQueryFeeModels(sequelize)
   sequelize = await sequelize.sync({ force: true })
 
@@ -49,6 +51,7 @@ const setup = async () => {
   const network = await Network.create(
     logger,
     testNetworkSpecification,
+    models,
     queryFeeModels,
     graphNode,
     metrics,
diff --git a/packages/indexer-common/src/allocations/escrow-accounts.ts b/packages/indexer-common/src/allocations/escrow-accounts.ts
index 1d126fd94..45402171c 100644
--- a/packages/indexer-common/src/allocations/escrow-accounts.ts
+++ b/packages/indexer-common/src/allocations/escrow-accounts.ts
@@ -13,6 +13,14 @@ export type EscrowAccountResponse = {
   }[]
 }
 
+export type EscrowSenderResponse = {
+  signer: {
+    sender: {
+      id: string
+    }
+  }
+}
+
 export class EscrowAccounts {
   constructor(private sendersBalances: Map<Address, U256>) {}
 
@@ -65,3 +73,26 @@ export const getEscrowAccounts = async (
   }
   return EscrowAccounts.fromResponse(result.data)
 }
+
+export const getEscrowSenderForSigner = async (
+  tapSubgraph: SubgraphClient,
+  signer: Address,
+): Promise<Address> => {
+  const signerLower = signer.toLowerCase()
+  const result = await tapSubgraph.query<EscrowSenderResponse>(
+    gql`
+      query EscrowAccountQuery($signer: ID!) {
+        signer(id: $signer) {
+          sender {
+            id
+          }
+        }
+      }
+    `,
+    { signer: signerLower },
+  )
+  if (!result.data) {
+    throw `There was an error while querying Tap Subgraph. Errors: ${result.error}`
+  }
+  return toAddress(result.data.signer.sender.id)
+}
diff --git a/packages/indexer-common/src/graph-node.ts b/packages/indexer-common/src/graph-node.ts
index 494247e44..0aca42720 100644
--- a/packages/indexer-common/src/graph-node.ts
+++ b/packages/indexer-common/src/graph-node.ts
@@ -708,6 +708,22 @@ export class GraphNode {
     }
   }
 
+  public async entityCount(deployments: SubgraphDeploymentID[]): Promise<number[]> {
+    // Query the entity count for each deployment using the indexingStatuses query
+    const query = `
+      query entityCounts($deployments: [String!]!) {
+          indexingStatuses(subgraphs: $deployments) {
+            entityCount
+          }
+        }
+    `
+    const result = await this.status
+      .query(query, { deployments: deployments.map((id) => id.ipfsHash) })
+      .toPromise()
+
+    return result.data.indexingStatuses.map((status) => status.entityCount) as number[]
+  }
+
   public async proofOfIndexing(
     deployment: SubgraphDeploymentID,
     block: BlockPointer,
diff --git a/packages/indexer-common/src/index.ts b/packages/indexer-common/src/index.ts
index e22722403..09d6efe05 100644
--- a/packages/indexer-common/src/index.ts
+++ b/packages/indexer-common/src/index.ts
@@ -3,6 +3,7 @@ export * from './allocations'
 export * from './async-cache'
 export * from './errors'
 export * from './indexer-management'
+export * from './indexing-fees'
 export * from './graph-node'
 export * from './operator'
 export * from './network'
@@ -16,3 +17,4 @@ export * from './utils'
 export * from './parsers'
 export * as specification from './network-specification'
 export * from './sequential-timer'
+export * from './indexing-fees'
diff --git a/packages/indexer-common/src/indexer-management/__tests__/allocations.test.ts b/packages/indexer-common/src/indexer-management/__tests__/allocations.test.ts
index 5c3a0ad28..d7569f8d8 100644
--- a/packages/indexer-common/src/indexer-management/__tests__/allocations.test.ts
+++ b/packages/indexer-common/src/indexer-management/__tests__/allocations.test.ts
@@ -61,6 +61,7 @@ const setup = async () => {
   const network = await Network.create(
     logger,
     testNetworkSpecification,
+    managementModels,
     queryFeeModels,
     graphNode,
     metrics,
diff --git a/packages/indexer-common/src/indexer-management/__tests__/util.ts b/packages/indexer-common/src/indexer-management/__tests__/util.ts
index 353cadb4e..328e7a15e 100644
--- a/packages/indexer-common/src/indexer-management/__tests__/util.ts
+++ b/packages/indexer-common/src/indexer-management/__tests__/util.ts
@@ -56,6 +56,7 @@ export const createTestManagementClient = async (
   const network = await Network.create(
     logger,
     networkSpecification,
+    managementModels,
     queryFeeModels,
     graphNode,
     metrics,
diff --git a/packages/indexer-common/src/indexer-management/allocations.ts b/packages/indexer-common/src/indexer-management/allocations.ts
index ee7c3fe58..eb37b767d 100644
--- a/packages/indexer-common/src/indexer-management/allocations.ts
+++ b/packages/indexer-common/src/indexer-management/allocations.ts
@@ -15,6 +15,7 @@ import {
   AllocationStatus,
   CloseAllocationResult,
   CreateAllocationResult,
+  DipsManager,
   fetchIndexingRules,
   GraphNode,
   indexerError,
@@ -98,12 +99,17 @@ export type TransactionResult =
   | ActionFailure[]
 
 export class AllocationManager {
+  declare dipsManager: DipsManager | null
   constructor(
     private logger: Logger,
     private models: IndexerManagementModels,
     private graphNode: GraphNode,
     private network: Network,
-  ) {}
+  ) {
+    if (this.network.specification.indexerOptions.dipperEndpoint) {
+      this.dipsManager = new DipsManager(this.logger, this.models, this.network, this)
+    }
+  }
 
   async executeBatch(actions: Action[]): Promise<AllocationResult[]> {
     const logger = this.logger.child({ function: 'executeBatch' })
@@ -511,6 +517,14 @@ export class AllocationManager {
       await upsertIndexingRule(logger, this.models, indexingRule)
     }
 
+    if (this.dipsManager) {
+      await this.dipsManager.tryUpdateAgreementAllocation(
+        deployment,
+        null,
+        toAddress(createAllocationEventLogs.allocationID),
+      )
+    }
+
     return {
       actionID,
       type: 'allocate',
@@ -667,6 +681,15 @@ export class AllocationManager {
 
     await upsertIndexingRule(logger, this.models, neverIndexingRule)
 
+    if (this.dipsManager) {
+      await this.dipsManager.tryCancelAgreement(allocationID)
+      await this.dipsManager.tryUpdateAgreementAllocation(
+        allocation.subgraphDeployment.id.toString(),
+        toAddress(allocationID),
+        null,
+      )
+    }
+
     return {
       actionID,
       type: 'unallocate',
@@ -966,6 +989,14 @@ export class AllocationManager {
       await upsertIndexingRule(logger, this.models, indexingRule)
     }
 
+    if (this.dipsManager) {
+      await this.dipsManager.tryUpdateAgreementAllocation(
+        subgraphDeploymentID.toString(),
+        toAddress(allocationID),
+        toAddress(createAllocationEventLogs.allocationID),
+      )
+    }
+
     return {
       actionID,
       type: 'reallocate',
diff --git a/packages/indexer-common/src/indexer-management/client.ts b/packages/indexer-common/src/indexer-management/client.ts
index 5db42b2dd..4b77c2073 100644
--- a/packages/indexer-common/src/indexer-management/client.ts
+++ b/packages/indexer-common/src/indexer-management/client.ts
@@ -452,9 +452,11 @@ export interface IndexerManagementClientOptions {
   graphNode: GraphNode
   network: Network
   defaults: IndexerManagementDefaults
+  actionManager?: ActionManager | undefined
 }
 
 export class IndexerManagementClient extends Client {
+  declare actionManager: ActionManager | undefined
   private logger?: Logger
   private models: IndexerManagementModels
 
@@ -463,6 +465,7 @@ export class IndexerManagementClient extends Client {
 
     this.logger = options.logger
     this.models = options.models
+    this.actionManager = options.actionManager
   }
 }
 
@@ -499,5 +502,8 @@ export const createIndexerManagementClient = async (
     context,
   })
 
-  return new IndexerManagementClient({ url: 'no-op', exchanges: [exchange] }, options)
+  return new IndexerManagementClient(
+    { url: 'no-op', exchanges: [exchange] },
+    { ...options, actionManager },
+  )
 }
diff --git a/packages/indexer-common/src/indexer-management/models/index.ts b/packages/indexer-common/src/indexer-management/models/index.ts
index 8d5ec55af..81a59f4d3 100644
--- a/packages/indexer-common/src/indexer-management/models/index.ts
+++ b/packages/indexer-common/src/indexer-management/models/index.ts
@@ -4,6 +4,7 @@ import { IndexingRuleModels, defineIndexingRuleModels } from './indexing-rule'
 import { CostModelModels, defineCostModelModels } from './cost-model'
 import { POIDisputeModels, definePOIDisputeModels } from './poi-dispute'
 import { ActionModels, defineActionModels } from './action'
+import { defineIndexingFeesModels, IndexingFeesModels } from './indexing-agreement'
 
 export * from './cost-model'
 export * from './indexing-rule'
@@ -13,7 +14,8 @@ export * from './action'
 export type IndexerManagementModels = IndexingRuleModels &
   CostModelModels &
   POIDisputeModels &
-  ActionModels
+  ActionModels &
+  IndexingFeesModels
 
 export const defineIndexerManagementModels = (
   sequelize: Sequelize,
@@ -24,4 +26,5 @@ export const defineIndexerManagementModels = (
     defineIndexingRuleModels(sequelize),
     definePOIDisputeModels(sequelize),
     defineActionModels(sequelize),
+    defineIndexingFeesModels(sequelize),
   )
diff --git a/packages/indexer-common/src/indexer-management/models/indexing-agreement.ts b/packages/indexer-common/src/indexer-management/models/indexing-agreement.ts
new file mode 100644
index 000000000..462456a7c
--- /dev/null
+++ b/packages/indexer-common/src/indexer-management/models/indexing-agreement.ts
@@ -0,0 +1,215 @@
+import { toAddress, Address } from '@graphprotocol/common-ts'
+import {
+  DataTypes,
+  Sequelize,
+  Model,
+  CreationOptional,
+  InferCreationAttributes,
+  InferAttributes,
+} from 'sequelize'
+
+// Indexing Fees AKA "DIPs"
+
+export class IndexingAgreement extends Model<
+  InferAttributes<IndexingAgreement>,
+  InferCreationAttributes<IndexingAgreement>
+> {
+  declare id: CreationOptional<string>
+  declare signature: Buffer
+  declare signed_payload: Buffer
+  declare protocol_network: string
+  declare chain_id: string
+  declare base_price_per_epoch: string
+  declare price_per_entity: string
+  declare subgraph_deployment_id: string
+  declare service: string
+  declare payee: string
+  declare payer: string
+  declare deadline: Date
+  declare duration_epochs: bigint
+  declare max_initial_amount: string
+  declare max_ongoing_amount_per_epoch: string
+  declare min_epochs_per_collection: bigint
+  declare max_epochs_per_collection: bigint
+  declare created_at: Date
+  declare updated_at: Date
+  declare cancelled_at: Date | null
+  declare signed_cancellation_payload: Buffer | null
+  declare current_allocation_id: string | null
+  declare last_allocation_id: string | null
+  declare last_payment_collected_at: Date | null
+}
+
+export interface IndexingFeesModels {
+  IndexingAgreement: typeof IndexingAgreement
+}
+
+export const defineIndexingFeesModels = (sequelize: Sequelize): IndexingFeesModels => {
+  IndexingAgreement.init(
+    {
+      id: {
+        type: DataTypes.UUID,
+        primaryKey: true,
+      },
+      signature: {
+        type: DataTypes.BLOB,
+        allowNull: false,
+        unique: true,
+      },
+      signed_payload: {
+        type: DataTypes.BLOB,
+        allowNull: false,
+      },
+      protocol_network: {
+        type: DataTypes.STRING(255),
+        allowNull: false,
+      },
+      chain_id: {
+        type: DataTypes.STRING(255),
+        allowNull: false,
+      },
+      base_price_per_epoch: {
+        type: DataTypes.DECIMAL(39),
+        allowNull: false,
+      },
+      price_per_entity: {
+        type: DataTypes.DECIMAL(39),
+        allowNull: false,
+      },
+      subgraph_deployment_id: {
+        type: DataTypes.STRING(255),
+        allowNull: false,
+      },
+      service: {
+        type: DataTypes.CHAR(40),
+        allowNull: false,
+        get() {
+          const rawValue = this.getDataValue('service')
+          return toAddress(rawValue)
+        },
+        set(value: Address) {
+          const addressWithoutPrefix = value.toLowerCase().replace('0x', '')
+          this.setDataValue('service', addressWithoutPrefix)
+        },
+      },
+      payee: {
+        type: DataTypes.CHAR(40),
+        allowNull: false,
+        get() {
+          const rawValue = this.getDataValue('payee')
+          return toAddress(rawValue)
+        },
+        set(value: Address) {
+          const addressWithoutPrefix = value.toLowerCase().replace('0x', '')
+          this.setDataValue('payee', addressWithoutPrefix)
+        },
+      },
+      payer: {
+        type: DataTypes.CHAR(40),
+        allowNull: false,
+        get() {
+          const rawValue = this.getDataValue('payer')
+          return toAddress(rawValue)
+        },
+        set(value: Address) {
+          const addressWithoutPrefix = value.toLowerCase().replace('0x', '')
+          this.setDataValue('payer', addressWithoutPrefix)
+        },
+      },
+      deadline: {
+        type: DataTypes.DATE,
+        allowNull: false,
+      },
+      duration_epochs: {
+        type: DataTypes.BIGINT,
+        allowNull: false,
+      },
+      max_initial_amount: {
+        type: DataTypes.DECIMAL(39),
+        allowNull: false,
+      },
+      max_ongoing_amount_per_epoch: {
+        type: DataTypes.DECIMAL(39),
+        allowNull: false,
+      },
+      min_epochs_per_collection: {
+        type: DataTypes.BIGINT,
+        allowNull: false,
+      },
+      max_epochs_per_collection: {
+        type: DataTypes.BIGINT,
+        allowNull: false,
+      },
+      created_at: {
+        type: DataTypes.DATE,
+        allowNull: false,
+      },
+      updated_at: {
+        type: DataTypes.DATE,
+        allowNull: false,
+      },
+      cancelled_at: {
+        type: DataTypes.DATE,
+        allowNull: true,
+      },
+      signed_cancellation_payload: {
+        type: DataTypes.BLOB,
+        allowNull: true,
+      },
+      current_allocation_id: {
+        type: DataTypes.CHAR(40),
+        allowNull: true,
+        get() {
+          const rawValue = this.getDataValue('current_allocation_id')
+          if (!rawValue) {
+            return null
+          }
+          return toAddress(rawValue)
+        },
+        set(value: Address | null) {
+          if (!value) {
+            this.setDataValue('current_allocation_id', null)
+          } else {
+            const addressWithoutPrefix = value.toLowerCase().replace('0x', '')
+            this.setDataValue('current_allocation_id', addressWithoutPrefix)
+          }
+        },
+      },
+      last_allocation_id: {
+        type: DataTypes.CHAR(40),
+        allowNull: true,
+        get() {
+          const rawValue = this.getDataValue('last_allocation_id')
+          if (!rawValue) {
+            return null
+          }
+          return toAddress(rawValue)
+        },
+        set(value: Address | null) {
+          if (!value) {
+            this.setDataValue('last_allocation_id', null)
+          } else {
+            const addressWithoutPrefix = value.toLowerCase().replace('0x', '')
+            this.setDataValue('last_allocation_id', addressWithoutPrefix)
+          }
+        },
+      },
+      last_payment_collected_at: {
+        type: DataTypes.DATE,
+        allowNull: true,
+      },
+    },
+    {
+      modelName: 'IndexingAgreement',
+      sequelize,
+      tableName: 'indexing_agreements',
+      timestamps: true,
+      createdAt: 'created_at',
+      updatedAt: 'updated_at',
+    },
+  )
+
+  return {
+    ['IndexingAgreement']: IndexingAgreement,
+  }
+}
diff --git a/packages/indexer-common/src/indexer-management/resolvers/allocations.ts b/packages/indexer-common/src/indexer-management/resolvers/allocations.ts
index c614ac457..6c2d80856 100644
--- a/packages/indexer-common/src/indexer-management/resolvers/allocations.ts
+++ b/packages/indexer-common/src/indexer-management/resolvers/allocations.ts
@@ -359,7 +359,13 @@ export default {
       deployment: string
       amount: string
     },
-    { network, graphNode, logger, models }: IndexerManagementResolverContext,
+    {
+      network,
+      graphNode,
+      logger,
+      models,
+      actionManager,
+    }: IndexerManagementResolverContext,
   ): Promise<CreateAllocationResult> => {
     const protocolNetwork = network.specification.networkIdentifier
     logger.debug('Execute createAllocation() mutation', {
@@ -557,6 +563,14 @@ export default {
 
       await models.IndexingRule.upsert(indexingRule)
 
+      if (actionManager?.allocationManager?.dipsManager) {
+        await actionManager.allocationManager.dipsManager.tryUpdateAgreementAllocation(
+          deployment,
+          null,
+          toAddress(createAllocationEventLogs.allocationID),
+        )
+      }
+
       // Since upsert succeeded, we _must_ have a rule
       const updatedRule = await models.IndexingRule.findOne({
         where: { identifier: indexingRule.identifier },
@@ -594,7 +608,7 @@ export default {
       poi: string | undefined
       force: boolean
     },
-    { logger, models, network }: IndexerManagementResolverContext,
+    { logger, models, network, actionManager }: IndexerManagementResolverContext,
   ): Promise<CloseAllocationResult> => {
     logger.debug('Execute closeAllocation() mutation', {
       allocationID: allocation,
@@ -712,6 +726,15 @@ export default {
 
       await models.IndexingRule.upsert(offchainIndexingRule)
 
+      if (actionManager?.allocationManager?.dipsManager) {
+        await actionManager.allocationManager.dipsManager.tryCancelAgreement(allocation)
+        await actionManager.allocationManager.dipsManager.tryUpdateAgreementAllocation(
+          allocationData.subgraphDeployment.id.toString(),
+          toAddress(allocation),
+          null,
+        )
+      }
+
       // Since upsert succeeded, we _must_ have a rule
       const updatedRule = await models.IndexingRule.findOne({
         where: { identifier: offchainIndexingRule.identifier },
@@ -751,7 +774,7 @@ export default {
       force: boolean
       protocolNetwork: string
     },
-    { logger, models, network }: IndexerManagementResolverContext,
+    { logger, models, network, actionManager }: IndexerManagementResolverContext,
   ): Promise<ReallocateAllocationResult> => {
     logger = logger.child({
       component: 'reallocateAllocationResolver',
@@ -1018,6 +1041,14 @@ export default {
 
       await models.IndexingRule.upsert(indexingRule)
 
+      if (actionManager?.allocationManager?.dipsManager) {
+        await actionManager.allocationManager.dipsManager.tryUpdateAgreementAllocation(
+          allocationData.subgraphDeployment.id.toString(),
+          toAddress(allocation),
+          toAddress(createAllocationEventLogs.allocationID),
+        )
+      }
+
       // Since upsert succeeded, we _must_ have a rule
       const updatedRule = await models.IndexingRule.findOne({
         where: { identifier: indexingRule.identifier },
diff --git a/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts b/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts
new file mode 100644
index 000000000..204f8a884
--- /dev/null
+++ b/packages/indexer-common/src/indexing-fees/__tests__/dips.test.ts
@@ -0,0 +1,522 @@
+import {
+  DipsManager,
+  GraphNode,
+  IndexerManagementModels,
+  Network,
+  QueryFeeModels,
+  defineIndexerManagementModels,
+  defineQueryFeeModels,
+  SubgraphIdentifierType,
+  IndexingDecisionBasis,
+  AllocationManager,
+  DipsCollector,
+  TapCollector,
+  createIndexerManagementClient,
+  Operator,
+  ActionManager,
+  IndexerManagementClient,
+} from '@graphprotocol/indexer-common'
+import {
+  connectDatabase,
+  createLogger,
+  createMetrics,
+  Logger,
+  Metrics,
+  parseGRT,
+  SubgraphDeploymentID,
+  toAddress,
+} from '@graphprotocol/common-ts'
+import { Sequelize } from 'sequelize'
+import { testNetworkSpecification } from '../../indexer-management/__tests__/util'
+import { BigNumber } from 'ethers'
+import { CollectPaymentStatus } from '@graphprotocol/dips-proto/generated/gateway'
+
+// Make global Jest variables available
+// eslint-disable-next-line @typescript-eslint/no-explicit-any
+declare const __DATABASE__: any
+declare const __LOG_LEVEL__: never
+
+// Add these type declarations after the existing imports
+let sequelize: Sequelize
+let logger: Logger
+let metrics: Metrics
+let graphNode: GraphNode
+let managementModels: IndexerManagementModels
+let queryFeeModels: QueryFeeModels
+let network: Network
+let dipsCollector: DipsCollector
+let indexerManagementClient: IndexerManagementClient
+let operator: Operator
+const networkSpecWithDips = {
+  ...testNetworkSpecification,
+  indexerOptions: {
+    ...testNetworkSpecification.indexerOptions,
+    enableDips: true,
+    dipperEndpoint: 'https://test-dipper-endpoint.xyz',
+    dipsAllocationAmount: parseGRT('1.0'), // Amount of GRT to allocate for DIPs
+    dipsEpochsMargin: 1, // Optional: Number of epochs margin for DIPs
+  },
+}
+
+const mockSubgraphDeployment = (id: string) => {
+  return {
+    id: new SubgraphDeploymentID(id),
+    ipfsHash: id,
+    deniedAt: null,
+    stakedTokens: BigNumber.from('1000'),
+    signalledTokens: BigNumber.from('1000'),
+    queryFeesAmount: BigNumber.from('0'),
+    protocolNetwork: 'eip155:421614',
+  }
+}
+
+jest.spyOn(TapCollector.prototype, 'startRAVProcessing').mockImplementation(() => {})
+const startCollectionLoop = jest
+  .spyOn(DipsCollector.prototype, 'startCollectionLoop')
+  .mockImplementation(() => {})
+jest.spyOn(ActionManager.prototype, 'monitorQueue').mockImplementation(async () => {})
+const setup = async () => {
+  logger = createLogger({
+    name: 'DIPs Test Logger',
+    async: false,
+    level: __LOG_LEVEL__ ?? 'error',
+  })
+  metrics = createMetrics()
+  // Clearing the registry prevents duplicate metric registration in the default registry.
+  metrics.registry.clear()
+
+  graphNode = new GraphNode(
+    logger,
+    'https://test-admin-endpoint.xyz',
+    'https://test-query-endpoint.xyz',
+    'https://test-status-endpoint.xyz',
+  )
+
+  sequelize = await connectDatabase(__DATABASE__)
+  managementModels = defineIndexerManagementModels(sequelize)
+  queryFeeModels = defineQueryFeeModels(sequelize)
+  sequelize = await sequelize.sync({ force: true })
+
+  network = await Network.create(
+    logger,
+    networkSpecWithDips,
+    managementModels,
+    queryFeeModels,
+    graphNode,
+    metrics,
+  )
+  dipsCollector = network.dipsCollector!
+  indexerManagementClient = await createIndexerManagementClient({
+    models: managementModels,
+    graphNode,
+    logger,
+    defaults: {
+      globalIndexingRule: {
+        allocationAmount: parseGRT('1000'),
+        parallelAllocations: 1,
+      },
+    },
+    network,
+  })
+
+  operator = new Operator(logger, indexerManagementClient, networkSpecWithDips)
+}
+
+const ensureGlobalIndexingRule = async () => {
+  await operator.ensureGlobalIndexingRule()
+  logger.debug('Ensured global indexing rule')
+}
+
+const setupEach = async () => {
+  sequelize = await sequelize.sync({ force: true })
+}
+
+const teardownEach = async () => {
+  // Clear out query fee model tables
+  await queryFeeModels.allocationReceipts.truncate({ cascade: true })
+  await queryFeeModels.vouchers.truncate({ cascade: true })
+  await queryFeeModels.transferReceipts.truncate({ cascade: true })
+  await queryFeeModels.transfers.truncate({ cascade: true })
+  await queryFeeModels.allocationSummaries.truncate({ cascade: true })
+  await queryFeeModels.scalarTapReceipts.truncate({ cascade: true })
+
+  // Clear out indexer management models
+  await managementModels.Action.truncate({ cascade: true })
+  await managementModels.CostModel.truncate({ cascade: true })
+  await managementModels.IndexingRule.truncate({ cascade: true })
+  await managementModels.POIDispute.truncate({ cascade: true })
+
+  // Clear out indexing agreement model
+  await managementModels.IndexingAgreement.truncate({ cascade: true })
+}
+
+const teardownAll = async () => {
+  await sequelize.drop({})
+}
+
+describe('DipsManager', () => {
+  beforeAll(setup)
+  beforeEach(setupEach)
+  afterEach(teardownEach)
+  afterAll(teardownAll)
+
+  // We have been rate-limited on CI as this test uses RPC providers,
+  // so we set its timeout to a higher value than usual.
+  jest.setTimeout(30_000)
+
+  describe('initialization', () => {
+    test('creates DipsManager when dipperEndpoint is configured', () => {
+      const dipsManager = new DipsManager(logger, managementModels, network, null)
+      expect(dipsManager).toBeDefined()
+    })
+
+    test('throws error when dipperEndpoint is not configured', async () => {
+      const specWithoutDipper = {
+        ...testNetworkSpecification,
+        indexerOptions: {
+          ...testNetworkSpecification.indexerOptions,
+          dipperEndpoint: undefined,
+        },
+      }
+
+      metrics.registry.clear()
+      const networkWithoutDipper = await Network.create(
+        logger,
+        specWithoutDipper,
+        managementModels,
+        queryFeeModels,
+        graphNode,
+        metrics,
+      )
+      expect(
+        () => new DipsManager(logger, managementModels, networkWithoutDipper, null),
+      ).toThrow('dipperEndpoint is not set')
+    })
+  })
+
+  describe('agreement management', () => {
+    let dipsManager: DipsManager
+    const testDeploymentId = 'QmTZ8ejXJxRo7vDBS4uwqBeGoxLSWbhaA7oXa1RvxunLy7'
+    const testAllocationId = 'abcd47df40c29949a75a6693c77834c00b8ad626'
+    const testAgreementId = '123e4567-e89b-12d3-a456-426614174000'
+
+    beforeEach(async () => {
+      // Clear mock calls between tests
+      jest.clearAllMocks()
+
+      const allocationManager = new AllocationManager(
+        logger,
+        managementModels,
+        graphNode,
+        network,
+      )
+
+      dipsManager = new DipsManager(logger, managementModels, network, allocationManager)
+
+      // Create a test agreement
+      await managementModels.IndexingAgreement.create({
+        id: testAgreementId,
+        subgraph_deployment_id: testDeploymentId,
+        current_allocation_id: testAllocationId,
+        last_allocation_id: null,
+        last_payment_collected_at: null,
+        cancelled_at: null,
+        min_epochs_per_collection: BigInt(1),
+        max_epochs_per_collection: BigInt(5),
+        payer: '123456df40c29949a75a6693c77834c00b8a5678',
+        signature: Buffer.from('1234', 'hex'),
+        signed_payload: Buffer.from('5678', 'hex'),
+        protocol_network: 'arbitrum-sepolia',
+        chain_id: 'eip155:1',
+        base_price_per_epoch: '100',
+        price_per_entity: '1',
+        service: 'deadbedf40c29949a75a2293c11834c00b8a1234',
+        payee: '1212564f40c29949a75a3423c11834c00b8aaaaa',
+        deadline: new Date(Date.now() + 86400000), // 1 day from now
+        duration_epochs: BigInt(10),
+        max_initial_amount: '1000',
+        max_ongoing_amount_per_epoch: '100',
+        created_at: new Date(),
+        updated_at: new Date(),
+        signed_cancellation_payload: null,
+      })
+    })
+
+    test('cancels agreement when allocation is closed', async () => {
+      const client = dipsManager.gatewayDipsServiceClient
+
+      client.CancelAgreement = jest.fn().mockResolvedValue({})
+
+      await dipsManager.tryCancelAgreement(testAllocationId)
+
+      // Verify the client was called with correct parameters
+      expect((client.CancelAgreement as jest.Mock).mock.calls.length).toBe(1)
+      // TODO: Check the signed cancellation payload
+      expect((client.CancelAgreement as jest.Mock).mock.calls[0][0]).toEqual({
+        version: 1,
+        signedCancellation: expect.any(Uint8Array),
+      })
+
+      const agreement = await managementModels.IndexingAgreement.findOne({
+        where: { id: testAgreementId },
+      })
+      expect(agreement?.cancelled_at).toBeDefined()
+    })
+
+    test('handles errors when cancelling agreement', async () => {
+      const client = dipsManager.gatewayDipsServiceClient
+      client.CancelAgreement = jest
+        .fn()
+        .mockRejectedValueOnce(new Error('Failed to cancel'))
+
+      await dipsManager.tryCancelAgreement(testAllocationId)
+
+      const agreement = await managementModels.IndexingAgreement.findOne({
+        where: { id: testAgreementId },
+      })
+      expect(agreement?.cancelled_at).toBeNull()
+    })
+
+    test('updates agreement allocation IDs during reallocation', async () => {
+      const newAllocationId = '5678bedf40c29945678a2293c15678c00b8a5678'
+
+      await dipsManager.tryUpdateAgreementAllocation(
+        testDeploymentId,
+        toAddress(testAllocationId),
+        toAddress(newAllocationId),
+      )
+
+      const agreement = await managementModels.IndexingAgreement.findOne({
+        where: { id: testAgreementId },
+      })
+      expect(agreement?.current_allocation_id).toBe(toAddress(newAllocationId))
+      expect(agreement?.last_allocation_id).toBe(toAddress(testAllocationId))
+      expect(agreement?.last_payment_collected_at).toBeNull()
+    })
+
+    test('creates indexing rules for active agreements', async () => {
+      await ensureGlobalIndexingRule()
+      // Mock fetch the subgraph deployment from the network subgraph
+      network.networkMonitor.subgraphDeployment = jest
+        .fn()
+        .mockResolvedValue(mockSubgraphDeployment(testDeploymentId))
+
+      await dipsManager.ensureAgreementRules()
+
+      const rules = await managementModels.IndexingRule.findAll({
+        where: {
+          identifier: testDeploymentId,
+        },
+      })
+
+      expect(rules).toHaveLength(1)
+      expect(rules[0]).toMatchObject({
+        identifier: testDeploymentId,
+        identifierType: SubgraphIdentifierType.DEPLOYMENT,
+        decisionBasis: IndexingDecisionBasis.ALWAYS,
+        allocationAmount:
+          network.specification.indexerOptions.dipsAllocationAmount.toString(),
+        autoRenewal: true,
+        allocationLifetime: 4, // max_epochs_per_collection - dipsEpochsMargin
+      })
+    })
+
+    test('does not create or modify an indexing rule if it already exists', async () => {
+      await ensureGlobalIndexingRule()
+      // Create an indexing rule with the same identifier
+      await managementModels.IndexingRule.create({
+        identifier: testDeploymentId,
+        identifierType: SubgraphIdentifierType.DEPLOYMENT,
+        decisionBasis: IndexingDecisionBasis.ALWAYS,
+        allocationLifetime: 16,
+        requireSupported: true,
+        safety: true,
+        protocolNetwork: 'eip155:421614',
+        allocationAmount: '1030',
+      })
+
+      // Mock fetch the subgraph deployment from the network subgraph
+      network.networkMonitor.subgraphDeployment = jest
+        .fn()
+        .mockResolvedValue(mockSubgraphDeployment(testDeploymentId))
+
+      await dipsManager.ensureAgreementRules()
+
+      const rules = await managementModels.IndexingRule.findAll({
+        where: { identifier: testDeploymentId },
+      })
+      expect(rules).toHaveLength(1)
+      expect(rules[0]).toMatchObject({
+        identifier: testDeploymentId,
+        identifierType: SubgraphIdentifierType.DEPLOYMENT,
+        decisionBasis: IndexingDecisionBasis.ALWAYS,
+        allocationLifetime: 16,
+        requireSupported: true,
+        safety: true,
+        protocolNetwork: 'eip155:421614',
+        allocationAmount: '1030',
+      })
+    })
+
+    test('returns active DIPs deployments', async () => {
+      const deployments = await dipsManager.getActiveDipsDeployments()
+
+      expect(deployments).toHaveLength(1)
+      expect(deployments[0].ipfsHash).toBe(testDeploymentId)
+    })
+  })
+})
+
+describe('DipsCollector', () => {
+  beforeAll(setup)
+  beforeEach(setupEach)
+  afterEach(teardownEach)
+  afterAll(teardownAll)
+
+  describe('initialization', () => {
+    test('creates DipsCollector when dipperEndpoint is configured', () => {
+      const dipsCollector = new DipsCollector(
+        logger,
+        managementModels,
+        queryFeeModels,
+        networkSpecWithDips,
+        network.tapCollector!,
+        network.wallet,
+        graphNode,
+        jest.fn(),
+      )
+      expect(dipsCollector).toBeDefined()
+    })
+    test('starts payment collection loop', () => {
+      const dipsCollector = new DipsCollector(
+        logger,
+        managementModels,
+        queryFeeModels,
+        networkSpecWithDips,
+        network.tapCollector!,
+        network.wallet,
+        graphNode,
+        jest.fn(),
+      )
+      expect(dipsCollector).toBeDefined()
+      expect(startCollectionLoop).toHaveBeenCalled()
+    })
+    test('throws error when dipperEndpoint is not configured', () => {
+      const specWithoutDipper = {
+        ...testNetworkSpecification,
+        indexerOptions: {
+          ...testNetworkSpecification.indexerOptions,
+          dipperEndpoint: undefined,
+        },
+      }
+      expect(
+        () =>
+          new DipsCollector(
+            logger,
+            managementModels,
+            queryFeeModels,
+            specWithoutDipper,
+            network.tapCollector!,
+            network.wallet,
+            graphNode,
+            jest.fn(),
+          ),
+      ).toThrow('dipperEndpoint is not set')
+    })
+  })
+
+  describe('payment collection', () => {
+    const testDeploymentId = 'QmTZ8ejXJxRo7vDBS4uwqBeGoxLSWbhaA7oXa1RvxunLy7'
+    const testAllocationId = 'abcd47df40c29949a75a6693c77834c00b8ad626'
+    const testAgreementId = '123e4567-e89b-12d3-a456-426614174000'
+
+    beforeEach(async () => {
+      // Clear mock calls between tests
+      jest.clearAllMocks()
+
+      // Create a test agreement
+      // Note last_allocation_id is set to the testAllocationId
+      // current_allocation_id is set to null so that we can collect payment
+      // (also last_payment_collected_at is set to null)
+      await managementModels.IndexingAgreement.create({
+        id: testAgreementId,
+        subgraph_deployment_id: testDeploymentId,
+        current_allocation_id: null,
+        last_allocation_id: testAllocationId,
+        last_payment_collected_at: null,
+        cancelled_at: null,
+        min_epochs_per_collection: BigInt(1),
+        max_epochs_per_collection: BigInt(5),
+        payer: '123456df40c29949a75a6693c77834c00b8a5678',
+        signature: Buffer.from('1234', 'hex'),
+        signed_payload: Buffer.from('5678', 'hex'),
+        protocol_network: 'arbitrum-sepolia',
+        chain_id: 'eip155:1',
+        base_price_per_epoch: '100',
+        price_per_entity: '1',
+        service: 'deadbedf40c29949a75a2293c11834c00b8a1234',
+        payee: '1212564f40c29949a75a3423c11834c00b8aaaaa',
+        deadline: new Date(Date.now() + 86400000), // 1 day from now
+        duration_epochs: BigInt(10),
+        max_initial_amount: '1000',
+        max_ongoing_amount_per_epoch: '100',
+        created_at: new Date(),
+        updated_at: new Date(),
+        signed_cancellation_payload: null,
+      })
+      graphNode.entityCount = jest.fn().mockResolvedValue([250000])
+    })
+    test('collects payment for a specific agreement', async () => {
+      const agreement = await managementModels.IndexingAgreement.findOne({
+        where: { id: testAgreementId },
+      })
+      if (!agreement) {
+        throw new Error('Agreement not found')
+      }
+
+      const client = dipsCollector.gatewayDipsServiceClient
+
+      client.CollectPayment = jest.fn().mockResolvedValue({
+        version: 1,
+        status: CollectPaymentStatus.ACCEPT,
+        tapReceipt: Buffer.from('1234', 'hex'),
+      })
+      dipsCollector.gatewayDipsServiceMessagesCodec.decodeTapReceipt = jest
+        .fn()
+        .mockImplementation(() => {
+          logger.info('MOCK Decoding TAP receipt')
+          return {
+            allocation_id: toAddress(testAllocationId),
+            signer_address: toAddress('0xabcd56df41234949a75a6693c77834c00b8abbbb'),
+            signature: Buffer.from('1234', 'hex'),
+            timestamp_ns: 1234567890,
+            nonce: 1,
+            value: '1000',
+          }
+        })
+      dipsCollector.escrowSenderGetter = jest.fn().mockImplementation(() => {
+        logger.info('MOCK Getting escrow sender for signer')
+        return toAddress('0x123456df40c29949a75a6693c77834c00b8a5678')
+      })
+
+      await dipsCollector.tryCollectPayment(agreement)
+
+      expect(client.CollectPayment).toHaveBeenCalledWith({
+        version: 1,
+        signedCollection: expect.any(Uint8Array),
+      })
+      expect(agreement.last_payment_collected_at).not.toBeNull()
+
+      const receipt = await queryFeeModels.scalarTapReceipts.findOne({
+        where: {
+          allocation_id: testAllocationId,
+        },
+      })
+      expect(receipt).not.toBeNull()
+      expect(receipt?.signer_address).toBe(
+        toAddress('0xabcd56df41234949a75a6693c77834c00b8abbbb'),
+      )
+      expect(receipt?.value).toBe('1000')
+    })
+  })
+})
diff --git a/packages/indexer-common/src/indexing-fees/dips.ts b/packages/indexer-common/src/indexing-fees/dips.ts
new file mode 100644
index 000000000..0eede21ea
--- /dev/null
+++ b/packages/indexer-common/src/indexing-fees/dips.ts
@@ -0,0 +1,327 @@
+import {
+  Address,
+  formatGRT,
+  Logger,
+  SubgraphDeploymentID,
+  toAddress,
+} from '@graphprotocol/common-ts'
+import {
+  AllocationManager,
+  getEscrowSenderForSigner,
+  GraphNode,
+  IndexerManagementModels,
+  IndexingDecisionBasis,
+  IndexingRuleAttributes,
+  Network,
+  QueryFeeModels,
+  sequentialTimerMap,
+  SubgraphClient,
+  SubgraphIdentifierType,
+  TapCollector,
+  upsertIndexingRule,
+} from '@graphprotocol/indexer-common'
+import { Op } from 'sequelize'
+
+import {
+  createGatewayDipsServiceClient,
+  GatewayDipsServiceMessagesCodec,
+} from './gateway-dips-service-client'
+import {
+  CollectPaymentStatus,
+  GatewayDipsServiceClientImpl,
+} from '@graphprotocol/dips-proto/generated/gateway'
+import { IndexingAgreement } from '../indexer-management/models/indexing-agreement'
+import { NetworkSpecification } from '../network-specification'
+import { Wallet } from 'ethers'
+
+const DIPS_COLLECTION_INTERVAL = 60_000
+
+const uuidToHex = (uuid: string) => {
+  return `0x${uuid.replace(/-/g, '')}`
+}
+
+const normalizeAddressForDB = (address: string) => {
+  return toAddress(address).toLowerCase().replace('0x', '')
+}
+
+type GetEscrowSenderForSigner = (
+  tapSubgraph: SubgraphClient,
+  signer: Address,
+) => Promise<Address>
+export class DipsManager {
+  declare gatewayDipsServiceClient: GatewayDipsServiceClientImpl
+  declare gatewayDipsServiceMessagesCodec: GatewayDipsServiceMessagesCodec
+  constructor(
+    private logger: Logger,
+    private models: IndexerManagementModels,
+    private network: Network,
+    private parent: AllocationManager | null,
+  ) {
+    if (!this.network.specification.indexerOptions.dipperEndpoint) {
+      throw new Error('dipperEndpoint is not set')
+    }
+    this.gatewayDipsServiceClient = createGatewayDipsServiceClient(
+      this.network.specification.indexerOptions.dipperEndpoint,
+    )
+    this.gatewayDipsServiceMessagesCodec = new GatewayDipsServiceMessagesCodec()
+  }
+  // Cancel an agreement associated to an allocation if it exists
+  async tryCancelAgreement(allocationId: string) {
+    const normalizedAllocationId = normalizeAddressForDB(allocationId)
+    const agreement = await this.models.IndexingAgreement.findOne({
+      where: {
+        current_allocation_id: normalizedAllocationId,
+        cancelled_at: null,
+      },
+    })
+    if (agreement) {
+      try {
+        const cancellation =
+          await this.gatewayDipsServiceMessagesCodec.createSignedCancellationRequest(
+            uuidToHex(agreement.id),
+            this.network.wallet,
+          )
+        await this.gatewayDipsServiceClient.CancelAgreement({
+          version: 1,
+          signedCancellation: cancellation,
+        })
+
+        // Mark the agreement as cancelled
+        agreement.cancelled_at = new Date()
+        agreement.updated_at = new Date()
+        await agreement.save()
+      } catch (error) {
+        this.logger.error(`Error cancelling agreement ${agreement.id}`, { error })
+      }
+    }
+  }
+  // Update the current and last allocation ids for an agreement if it exists
+  async tryUpdateAgreementAllocation(
+    deploymentId: string,
+    oldAllocationId: Address | null,
+    newAllocationId: Address | null,
+  ) {
+    const agreement = await this.models.IndexingAgreement.findOne({
+      where: {
+        subgraph_deployment_id: deploymentId,
+      },
+    })
+    if (agreement) {
+      agreement.current_allocation_id = newAllocationId
+      agreement.last_allocation_id = oldAllocationId
+      agreement.last_payment_collected_at = null
+      agreement.updated_at = new Date()
+      await agreement.save()
+    }
+  }
+  async ensureAgreementRules() {
+    if (!this.parent) {
+      this.logger.error(
+        'DipsManager has no parent AllocationManager, cannot ensure agreement rules',
+      )
+      return
+    }
+    // Get all the indexing agreements that are not cancelled
+    const indexingAgreements = await this.models.IndexingAgreement.findAll({
+      where: {
+        cancelled_at: null,
+      },
+    })
+    // For each agreement, check that there is an indexing rule to always
+    // allocate to the agreement's subgraphDeploymentId, and if not, create one
+    for (const agreement of indexingAgreements) {
+      const subgraphDeploymentID = new SubgraphDeploymentID(
+        agreement.subgraph_deployment_id,
+      )
+      this.logger.info(
+        `Checking if indexing rule exists for agreement ${
+          agreement.id
+        }, deployment ${subgraphDeploymentID.toString()}`,
+      )
+      // If there is not yet an indexingRule that deems this deployment worth allocating to, make one
+      const ruleExists = await this.parent.matchingRuleExists(
+        this.logger,
+        subgraphDeploymentID,
+      )
+      if (!ruleExists) {
+        this.logger.info(
+          `Creating indexing rule for agreement ${agreement.id}, deployment ${agreement.subgraph_deployment_id}`,
+        )
+        const indexingRule = {
+          identifier: agreement.subgraph_deployment_id,
+          allocationAmount: formatGRT(
+            this.network.specification.indexerOptions.dipsAllocationAmount,
+          ),
+          identifierType: SubgraphIdentifierType.DEPLOYMENT,
+          decisionBasis: IndexingDecisionBasis.ALWAYS,
+          protocolNetwork: this.network.specification.networkIdentifier,
+          autoRenewal: true,
+          allocationLifetime: Math.max(
+            Number(agreement.min_epochs_per_collection),
+            Number(agreement.max_epochs_per_collection) -
+              this.network.specification.indexerOptions.dipsEpochsMargin,
+          ),
+        } as Partial<IndexingRuleAttributes>
+
+        await upsertIndexingRule(this.logger, this.models, indexingRule)
+      }
+    }
+  }
+  async getActiveDipsDeployments(): Promise<SubgraphDeploymentID[]> {
+    // Get all the indexing agreements that are not cancelled
+    const indexingAgreements = await this.models.IndexingAgreement.findAll({
+      where: {
+        cancelled_at: null,
+      },
+    })
+    return indexingAgreements.map(
+      (agreement) => new SubgraphDeploymentID(agreement.subgraph_deployment_id),
+    )
+  }
+}
+
+export class DipsCollector {
+  declare gatewayDipsServiceClient: GatewayDipsServiceClientImpl
+  declare gatewayDipsServiceMessagesCodec: GatewayDipsServiceMessagesCodec
+  constructor(
+    private logger: Logger,
+    private managementModels: IndexerManagementModels,
+    private queryFeeModels: QueryFeeModels,
+    private specification: NetworkSpecification,
+    private tapCollector: TapCollector,
+    private wallet: Wallet,
+    private graphNode: GraphNode,
+    public escrowSenderGetter: GetEscrowSenderForSigner,
+  ) {
+    if (!this.specification.indexerOptions.dipperEndpoint) {
+      throw new Error('dipperEndpoint is not set')
+    }
+    this.gatewayDipsServiceClient = createGatewayDipsServiceClient(
+      this.specification.indexerOptions.dipperEndpoint,
+    )
+    this.gatewayDipsServiceMessagesCodec = new GatewayDipsServiceMessagesCodec()
+  }
+
+  static create(
+    logger: Logger,
+    managementModels: IndexerManagementModels,
+    queryFeeModels: QueryFeeModels,
+    specification: NetworkSpecification,
+    tapCollector: TapCollector,
+    wallet: Wallet,
+    graphNode: GraphNode,
+    escrowSenderGetter?: GetEscrowSenderForSigner,
+  ) {
+    const collector = new DipsCollector(
+      logger,
+      managementModels,
+      queryFeeModels,
+      specification,
+      tapCollector,
+      wallet,
+      graphNode,
+      escrowSenderGetter ?? getEscrowSenderForSigner,
+    )
+    collector.startCollectionLoop()
+    return collector
+  }
+
+  startCollectionLoop() {
+    sequentialTimerMap(
+      {
+        logger: this.logger,
+        milliseconds: DIPS_COLLECTION_INTERVAL,
+      },
+      async () => {
+        this.logger.debug('Running DIPs payment collection loop')
+        await this.collectAllPayments()
+      },
+      {
+        onError: (err) => {
+          this.logger.error('Failed to collect DIPs payments', { err })
+        },
+      },
+    )
+  }
+
+  // Collect payments for all outstanding agreements
+  async collectAllPayments() {
+    const outstandingAgreements = await this.managementModels.IndexingAgreement.findAll({
+      where: {
+        last_payment_collected_at: null,
+        last_allocation_id: {
+          [Op.ne]: null,
+        },
+      },
+    })
+    for (const agreement of outstandingAgreements) {
+      await this.tryCollectPayment(agreement)
+    }
+  }
+  async tryCollectPayment(agreement: IndexingAgreement) {
+    if (!agreement.last_allocation_id) {
+      this.logger.error(`Agreement ${agreement.id} has no last allocation id`)
+      return
+    }
+    const entityCounts = await this.graphNode.entityCount([
+      new SubgraphDeploymentID(agreement.subgraph_deployment_id),
+    ])
+    if (entityCounts.length === 0) {
+      this.logger.error(`Agreement ${agreement.id} has no entity count`)
+      return
+    }
+    const entityCount = entityCounts[0]
+    const collection =
+      await this.gatewayDipsServiceMessagesCodec.createSignedCollectionRequest(
+        uuidToHex(agreement.id),
+        agreement.last_allocation_id,
+        entityCount,
+        this.wallet,
+      )
+    try {
+      this.logger.info(`Collecting payment for agreement ${agreement.id}`)
+      const response = await this.gatewayDipsServiceClient.CollectPayment({
+        version: 1,
+        signedCollection: collection,
+      })
+      if (response.status === CollectPaymentStatus.ACCEPT) {
+        if (!this.tapCollector) {
+          throw new Error('TapCollector not initialized')
+        }
+        // Store the tap receipt in the database
+        this.logger.info('Decoding TAP receipt for agreement')
+        const tapReceipt = this.gatewayDipsServiceMessagesCodec.decodeTapReceipt(
+          response.tapReceipt,
+          this.tapCollector?.tapContracts.tapVerifier.address,
+        )
+        // Check that the signer of the TAP receipt is a signer
+        // on the corresponding escrow account for the payer (sender) of the
+        // indexing agreement
+        const escrowSender = await this.escrowSenderGetter(
+          this.tapCollector?.tapSubgraph,
+          tapReceipt.signer_address,
+        )
+        if (escrowSender !== toAddress(agreement.payer)) {
+          // TODO: should we cancel the agreement here?
+          throw new Error(
+            'Signer of TAP receipt is not a signer on the indexing agreement',
+          )
+        }
+        if (tapReceipt.allocation_id !== toAddress(agreement.last_allocation_id)) {
+          throw new Error('Allocation ID mismatch')
+        }
+        await this.queryFeeModels.scalarTapReceipts.create(tapReceipt)
+        // Mark the agreement as having had a payment collected
+        agreement.last_payment_collected_at = new Date()
+        agreement.updated_at = new Date()
+        await agreement.save()
+      } else {
+        throw new Error(`Payment request not accepted: ${response.status}`)
+      }
+    } catch (error) {
+      this.logger.error(`Error collecting payment for agreement ${agreement.id}`, {
+        error,
+      })
+    }
+  }
+}
diff --git a/packages/indexer-common/src/indexing-fees/gateway-dips-service-client.ts b/packages/indexer-common/src/indexing-fees/gateway-dips-service-client.ts
new file mode 100644
index 000000000..1bfb832a5
--- /dev/null
+++ b/packages/indexer-common/src/indexing-fees/gateway-dips-service-client.ts
@@ -0,0 +1,168 @@
+import { Client, credentials } from '@grpc/grpc-js'
+import { UnaryCallback } from '@grpc/grpc-js/build/src/client'
+import { GatewayDipsServiceClientImpl } from '@graphprotocol/dips-proto/generated/gateway'
+import { Wallet } from 'ethers'
+import {
+  _TypedDataEncoder,
+  arrayify,
+  defaultAbiCoder,
+  recoverAddress,
+} from 'ethers/lib/utils'
+import { toAddress } from '@graphprotocol/common-ts'
+
+type RpcImpl = (service: string, method: string, data: Uint8Array) => Promise<Uint8Array>
+
+interface Rpc {
+  request: RpcImpl
+}
+
+export const domainSalt =
+  '0xb4632c657c26dce5d4d7da1d65bda185b14ff8f905ddbb03ea0382ed06c5ef28'
+export const chainId = 0xa4b1 // 42161
+export const cancelAgreementDomain = {
+  name: 'Graph Protocol Indexing Agreement Cancellation',
+  version: '0',
+  chainId: chainId,
+  salt: domainSalt,
+}
+export const cancelAgreementTypes = {
+  CancellationRequest: [{ name: 'agreement_id', type: 'bytes16' }],
+}
+
+export const collectPaymentsDomain = {
+  name: 'Graph Protocol Indexing Agreement Collection',
+  version: '0',
+  chainId: chainId,
+  salt: domainSalt,
+}
+export const collectPaymentsTypes = {
+  CollectionRequest: [
+    { name: 'agreement_id', type: 'bytes16' },
+    { name: 'allocation_id', type: 'address' },
+    { name: 'entity_count', type: 'uint64' },
+  ],
+}
+
+export class GatewayDipsServiceMessagesCodec {
+  async createSignedCancellationRequest(
+    agreementId: string,
+    wallet: Wallet,
+  ): Promise<Uint8Array> {
+    const signature = await wallet._signTypedData(
+      cancelAgreementDomain,
+      cancelAgreementTypes,
+      { agreement_id: agreementId },
+    )
+    return arrayify(
+      defaultAbiCoder.encode(['tuple(bytes16)', 'bytes'], [[agreementId], signature]),
+    )
+  }
+
+  async createSignedCollectionRequest(
+    agreementId: string,
+    allocationId: string,
+    entityCount: number,
+    wallet: Wallet,
+  ): Promise<Uint8Array> {
+    const signature = await wallet._signTypedData(
+      collectPaymentsDomain,
+      collectPaymentsTypes,
+      {
+        agreement_id: agreementId,
+        allocation_id: toAddress(allocationId),
+        entity_count: entityCount,
+      },
+    )
+    return arrayify(
+      defaultAbiCoder.encode(
+        ['tuple(bytes16, address, uint64)', 'bytes'],
+        [[agreementId, toAddress(allocationId), entityCount], signature],
+      ),
+    )
+  }
+
+  decodeTapReceipt(receipt: Uint8Array, verifyingContract: string) {
+    const [message, signature] = defaultAbiCoder.decode(
+      ['tuple(address,uint64,uint64,uint128)', 'bytes'],
+      receipt,
+    )
+
+    const [allocationId, timestampNs, nonce, value] = message
+
+    // Recover the signer address from the signature
+    // compute the EIP-712 digest of the message
+    const domain = {
+      name: 'TAP',
+      version: '1',
+      chainId: chainId,
+      verifyingContract,
+    }
+
+    const types = {
+      Receipt: [
+        { name: 'allocation_id', type: 'address' },
+        { name: 'timestamp_ns', type: 'uint64' },
+        { name: 'nonce', type: 'uint64' },
+        { name: 'value', type: 'uint128' },
+      ],
+    }
+
+    const digest = _TypedDataEncoder.hash(domain, types, {
+      allocation_id: allocationId,
+      timestamp_ns: timestampNs,
+      nonce: nonce,
+      value: value,
+    })
+    const signerAddress = recoverAddress(digest, signature)
+    return {
+      allocation_id: toAddress(allocationId),
+      signer_address: toAddress(signerAddress),
+      signature: signature,
+      timestamp_ns: timestampNs,
+      nonce: nonce,
+      value: value,
+    }
+  }
+}
+
+export const createRpc = (url: string): Rpc => {
+  const client = new Client(url, credentials.createInsecure())
+  const request: RpcImpl = (service, method, data) => {
+    // Conventionally in gRPC, the request path looks like
+    //   "package.names.ServiceName/MethodName",
+    // we therefore construct such a string
+    const path = `/${service}/${method}`
+
+    return new Promise((resolve, reject) => {
+      // makeUnaryRequest transmits the result (and error) with a callback
+      // transform this into a promise!
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      const resultCallback: UnaryCallback<any> = (err, res) => {
+        if (err) {
+          return reject(err)
+        }
+        resolve(res)
+      }
+      // eslint-disable-next-line @typescript-eslint/no-explicit-any
+      function passThrough(argument: any) {
+        return argument
+      }
+
+      // Using passThrough as the deserialize functions
+      client.makeUnaryRequest(
+        path,
+        (d) => Buffer.from(d),
+        passThrough,
+        data,
+        resultCallback,
+      )
+    })
+  }
+
+  return { request }
+}
+
+export const createGatewayDipsServiceClient = (url: string) => {
+  const rpc = createRpc(url)
+  return new GatewayDipsServiceClientImpl(rpc)
+}
diff --git a/packages/indexer-common/src/indexing-fees/index.ts b/packages/indexer-common/src/indexing-fees/index.ts
new file mode 100644
index 000000000..0b71f1b8e
--- /dev/null
+++ b/packages/indexer-common/src/indexing-fees/index.ts
@@ -0,0 +1 @@
+export * from './dips'
diff --git a/packages/indexer-common/src/network-specification.ts b/packages/indexer-common/src/network-specification.ts
index f683cdec5..4db7eb30b 100644
--- a/packages/indexer-common/src/network-specification.ts
+++ b/packages/indexer-common/src/network-specification.ts
@@ -58,6 +58,10 @@ export const IndexerOptions = z
     allocateOnNetworkSubgraph: z.boolean().default(false),
     register: z.boolean().default(true),
     finalityTime: positiveNumber().default(3600),
+    enableDips: z.boolean().default(false),
+    dipperEndpoint: z.string().url().optional(),
+    dipsAllocationAmount: GRT().default(1),
+    dipsEpochsMargin: positiveNumber().default(1),
   })
   .strict()
 export type IndexerOptions = z.infer<typeof IndexerOptions>
diff --git a/packages/indexer-common/src/network.ts b/packages/indexer-common/src/network.ts
index 1b8d436e2..053a23ce7 100644
--- a/packages/indexer-common/src/network.ts
+++ b/packages/indexer-common/src/network.ts
@@ -29,12 +29,14 @@ import {
   AllocationReceiptCollector,
   SubgraphFreshnessChecker,
   monitorEligibleAllocations,
+  IndexerManagementModels,
 } from '.'
 import { resolveChainId } from './indexer-management'
 import { monitorEthBalance } from './utils'
 import { QueryFeeModels } from './query-fees'
 import { readFileSync } from 'fs'
 import { TapCollector } from './allocations/tap-collector'
+import { DipsCollector } from './indexing-fees/dips'
 
 export class Network {
   logger: Logger
@@ -49,10 +51,12 @@ export class Network {
   receiptCollector: AllocationReceiptCollector | undefined
 
   tapCollector: TapCollector | undefined
+  dipsCollector: DipsCollector | undefined
   specification: spec.NetworkSpecification
   paused: Eventual<boolean>
   isOperator: Eventual<boolean>
-
+  queryFeeModels: QueryFeeModels
+  managementModels: IndexerManagementModels
   private constructor(
     logger: Logger,
     contracts: NetworkContracts,
@@ -66,6 +70,9 @@ export class Network {
     specification: spec.NetworkSpecification,
     paused: Eventual<boolean>,
     isOperator: Eventual<boolean>,
+    queryFeeModels: QueryFeeModels,
+    managementModels: IndexerManagementModels,
+    dipsCollector: DipsCollector | undefined,
   ) {
     this.logger = logger
     this.contracts = contracts
@@ -79,11 +86,15 @@ export class Network {
     this.specification = specification
     this.paused = paused
     this.isOperator = isOperator
+    this.queryFeeModels = queryFeeModels
+    this.managementModels = managementModels
+    this.dipsCollector = dipsCollector
   }
 
   static async create(
     parentLogger: Logger,
     specification: spec.NetworkSpecification,
+    managementModels: IndexerManagementModels,
     queryFeeModels: QueryFeeModels,
     graphNode: GraphNode,
     metrics: Metrics,
@@ -311,6 +322,7 @@ export class Network {
     // * TAP Collector
     // --------------------------------------------------------------------------------
     let tapCollector: TapCollector | undefined = undefined
+    let dipsCollector: DipsCollector | undefined = undefined
     if (tapContracts && tapSubgraph) {
       tapCollector = TapCollector.create({
         logger,
@@ -323,8 +335,19 @@ export class Network {
         tapSubgraph,
         networkSubgraph,
       })
+      if (specification.indexerOptions.enableDips) {
+        dipsCollector = DipsCollector.create(
+          logger,
+          managementModels,
+          queryFeeModels,
+          specification,
+          tapCollector,
+          wallet,
+          graphNode,
+        )
+      }
     } else {
-      logger.info(`RAV process not initiated. 
+      logger.info(`RAV (and DIPs) process not initiated. 
         Tap Contracts: ${!!tapContracts}. 
         Tap Subgraph: ${!!tapSubgraph}.`)
     }
@@ -345,6 +368,9 @@ export class Network {
       specification,
       paused,
       isOperator,
+      queryFeeModels,
+      managementModels,
+      dipsCollector,
     )
   }
 
diff --git a/packages/indexer-common/src/operator.ts b/packages/indexer-common/src/operator.ts
index 63d28fc52..9509ed19e 100644
--- a/packages/indexer-common/src/operator.ts
+++ b/packages/indexer-common/src/operator.ts
@@ -16,6 +16,7 @@ import {
   specification as spec,
   Action,
   POIDisputeAttributes,
+  DipsManager,
 } from '@graphprotocol/indexer-common'
 import { Logger, formatGRT } from '@graphprotocol/common-ts'
 import { BigNumber, utils } from 'ethers'
@@ -82,6 +83,10 @@ export class Operator {
     this.specification = specification
   }
 
+  get dipsManager(): DipsManager | null {
+    return this.indexerManagement.actionManager?.allocationManager?.dipsManager ?? null
+  }
+
   // --------------------------------------------------------------------------------
   // * Indexing Rules
   // --------------------------------------------------------------------------------
@@ -256,16 +261,26 @@ export class Operator {
     return result.data.actions
   }
 
-  async queueAction(action: ActionItem): Promise<Action[]> {
+  async queueAction(action: ActionItem, forceAction: boolean = false): Promise<Action[]> {
     let status = ActionStatus.QUEUED
     switch (this.specification.indexerOptions.allocationManagementMode) {
       case AllocationManagementMode.MANUAL:
-        throw Error(`Cannot queue actions when AllocationManagementMode = 'MANUAL'`)
+        if (forceAction) {
+          status = ActionStatus.APPROVED
+        } else {
+          throw Error(`Cannot queue actions when AllocationManagementMode = 'MANUAL'`)
+        }
+        break
       case AllocationManagementMode.AUTO:
         status = ActionStatus.APPROVED
         break
       case AllocationManagementMode.OVERSIGHT:
-        status = ActionStatus.QUEUED
+        if (forceAction) {
+          status = ActionStatus.APPROVED
+        } else {
+          status = ActionStatus.QUEUED
+        }
+        break
     }
 
     const actionInput = {
@@ -341,6 +356,7 @@ export class Operator {
     logger: Logger,
     deploymentAllocationDecision: AllocationDecision,
     mostRecentlyClosedAllocation: Allocation | undefined,
+    forceAction: boolean = false,
   ): Promise<void> {
     const desiredAllocationAmount = deploymentAllocationDecision.ruleMatch.rule
       ?.allocationAmount
@@ -369,14 +385,17 @@ export class Operator {
     }
 
     // Send AllocateAction to the queue
-    await this.queueAction({
-      params: {
-        deploymentID: deploymentAllocationDecision.deployment.ipfsHash,
-        amount: formatGRT(desiredAllocationAmount),
+    await this.queueAction(
+      {
+        params: {
+          deploymentID: deploymentAllocationDecision.deployment.ipfsHash,
+          amount: formatGRT(desiredAllocationAmount),
+        },
+        type: ActionType.ALLOCATE,
+        reason: deploymentAllocationDecision.reasonString(),
       },
-      type: ActionType.ALLOCATE,
-      reason: deploymentAllocationDecision.reasonString(),
-    })
+      forceAction,
+    )
 
     return
   }
@@ -385,6 +404,7 @@ export class Operator {
     logger: Logger,
     deploymentAllocationDecision: AllocationDecision,
     activeDeploymentAllocations: Allocation[],
+    forceAction: boolean = false,
   ): Promise<void> {
     const activeDeploymentAllocationsEligibleForClose = activeDeploymentAllocations.map(
       (allocation) => allocation.id,
@@ -404,16 +424,19 @@ export class Operator {
         activeDeploymentAllocationsEligibleForClose,
         async (allocation) => {
           // Send unallocate action to the queue
-          await this.queueAction({
-            params: {
-              allocationID: allocation,
-              deploymentID: deploymentAllocationDecision.deployment.ipfsHash,
-              poi: undefined,
-              force: false,
-            },
-            type: ActionType.UNALLOCATE,
-            reason: deploymentAllocationDecision.reasonString(),
-          } as ActionItem)
+          await this.queueAction(
+            {
+              params: {
+                allocationID: allocation,
+                deploymentID: deploymentAllocationDecision.deployment.ipfsHash,
+                poi: undefined,
+                force: false,
+              },
+              type: ActionType.UNALLOCATE,
+              reason: deploymentAllocationDecision.reasonString(),
+            } as ActionItem,
+            forceAction,
+          )
         },
         { concurrency: 1 },
       )
@@ -424,6 +447,7 @@ export class Operator {
     logger: Logger,
     deploymentAllocationDecision: AllocationDecision,
     expiredAllocations: Allocation[],
+    forceAction: boolean = false,
   ): Promise<void> {
     if (deploymentAllocationDecision.ruleMatch.rule?.autoRenewal) {
       logger.info(`Reallocating expired allocations`, {
@@ -440,15 +464,18 @@ export class Operator {
       await pMap(
         expiredAllocations,
         async (allocation) => {
-          await this.queueAction({
-            params: {
-              allocationID: allocation.id,
-              deploymentID: deploymentAllocationDecision.deployment.ipfsHash,
-              amount: formatGRT(desiredAllocationAmount),
+          await this.queueAction(
+            {
+              params: {
+                allocationID: allocation.id,
+                deploymentID: deploymentAllocationDecision.deployment.ipfsHash,
+                amount: formatGRT(desiredAllocationAmount),
+              },
+              type: ActionType.REALLOCATE,
+              reason: `${deploymentAllocationDecision.reasonString()}:allocationExpiring`, // Need to update to include 'ExpiringSoon'
             },
-            type: ActionType.REALLOCATE,
-            reason: `${deploymentAllocationDecision.reasonString()}:allocationExpiring`, // Need to update to include 'ExpiringSoon'
-          })
+            forceAction,
+          )
         },
         {
           stopOnError: false,
diff --git a/packages/indexer-common/src/query-fees/models.ts b/packages/indexer-common/src/query-fees/models.ts
index 095d60fe0..4b5d97d3b 100644
--- a/packages/indexer-common/src/query-fees/models.ts
+++ b/packages/indexer-common/src/query-fees/models.ts
@@ -5,19 +5,28 @@ import { TAPVerifier } from '@semiotic-labs/tap-contracts-bindings'
 
 export interface ScalarTapReceiptsAttributes {
   id: number
-  allocation_id: Address
-  signer_address: Address
+  allocation_id: string
+  signer_address: string
   signature: Uint8Array
   timestamp_ns: bigint
   nonce: bigint
   value: bigint
   error_log?: string
 }
+export interface ScalarTapReceiptsCreationAttributes {
+  allocation_id: string
+  signer_address: string
+  signature: Uint8Array
+  timestamp_ns: bigint
+  nonce: bigint
+  value: bigint
+}
+
 export class ScalarTapReceipts
-  extends Model<ScalarTapReceiptsAttributes>
+  extends Model<ScalarTapReceiptsAttributes, ScalarTapReceiptsCreationAttributes>
   implements ScalarTapReceiptsAttributes
 {
-  public id!: number
+  public id!: CreationOptional<number>
   public allocation_id!: Address
   public signer_address!: Address
   public signature!: Uint8Array
@@ -598,10 +607,26 @@ export function defineQueryFeeModels(sequelize: Sequelize): QueryFeeModels {
       allocation_id: {
         type: DataTypes.CHAR(40),
         allowNull: false,
+        get() {
+          const rawValue = this.getDataValue('allocation_id')
+          return toAddress(rawValue)
+        },
+        set(value: Address) {
+          const addressWithoutPrefix = value.toLowerCase().replace('0x', '')
+          this.setDataValue('allocation_id', addressWithoutPrefix)
+        },
       },
       signer_address: {
         type: DataTypes.CHAR(40),
         allowNull: false,
+        get() {
+          const rawValue = this.getDataValue('signer_address')
+          return toAddress(rawValue)
+        },
+        set(value: Address) {
+          const addressWithoutPrefix = value.toLowerCase().replace('0x', '')
+          this.setDataValue('signer_address', addressWithoutPrefix)
+        },
       },
       signature: {
         type: DataTypes.BLOB,
diff --git a/yarn.lock b/yarn.lock
index a48e38771..edd78420e 100644
--- a/yarn.lock
+++ b/yarn.lock
@@ -281,6 +281,11 @@
   resolved "https://registry.yarnpkg.com/@bcoe/v8-coverage/-/v8-coverage-0.2.3.tgz#75a2e8b51cb758a7553d6804a5932d7aace75c39"
   integrity sha512-0hYQ8SB4Db5zvZB4axdMHGwEaQjkZzFjQiN9LVYvIFB2nSUHW9tYpxWriPrWDASIxiaXax83REcLxuSdnGPZtw==
 
+"@bufbuild/protobuf@2.2.3", "@bufbuild/protobuf@^2.2.3":
+  version "2.2.3"
+  resolved "https://registry.yarnpkg.com/@bufbuild/protobuf/-/protobuf-2.2.3.tgz#9cd136f6b687e63e9b517b3a54211ece942897ee"
+  integrity sha512-tFQoXHJdkEOSwj5tRIZSPNUuXK3RaR7T1nUrPgbYX1pUbvqqaaZAsfo+NXBPsz5rZMSKVFrgK1WL8Q/MSLvprg==
+
 "@cspotcode/source-map-consumer@0.8.0":
   version "0.8.0"
   resolved "https://registry.yarnpkg.com/@cspotcode/source-map-consumer/-/source-map-consumer-0.8.0.tgz#33bf4b7b39c178821606f669bbc447a6a629786b"
@@ -968,6 +973,13 @@
     console-table-printer "^2.11.1"
     ethers "^5.6.0"
 
+"@graphprotocol/dips-proto@0.2.2":
+  version "0.2.2"
+  resolved "https://registry.yarnpkg.com/@graphprotocol/dips-proto/-/dips-proto-0.2.2.tgz#3beece3e768b8a7d64bace959e0bf15a91c3ef53"
+  integrity sha512-pAcnHnZ3qs2NrjYEUm8sahY0MBaV5KXfQVg9wk6f3LlClS1hK3a9aqUCI0CUriuALWbTwceeGgiKv8UIrJx4GA==
+  dependencies:
+    "@bufbuild/protobuf" "^2.2.3"
+
 "@graphprotocol/pino-sentry-simple@0.7.1":
   version "0.7.1"
   resolved "https://registry.yarnpkg.com/@graphprotocol/pino-sentry-simple/-/pino-sentry-simple-0.7.1.tgz#ac08b978bfa33178b9e809f53ae0983ff5f724d8"
@@ -978,6 +990,24 @@
     split2 "^3.1.1"
     through2 "^3.0.1"
 
+"@grpc/grpc-js@^1.12.6":
+  version "1.12.6"
+  resolved "https://registry.yarnpkg.com/@grpc/grpc-js/-/grpc-js-1.12.6.tgz#a3586ffdfb6a1f5cd5b4866dec9074c4a1e65472"
+  integrity sha512-JXUj6PI0oqqzTGvKtzOkxtpsyPRNsrmhh41TtIz/zEB6J+AUiZZ0dxWzcMwO9Ns5rmSPuMdghlTbUuqIM48d3Q==
+  dependencies:
+    "@grpc/proto-loader" "^0.7.13"
+    "@js-sdsl/ordered-map" "^4.4.2"
+
+"@grpc/proto-loader@^0.7.13":
+  version "0.7.13"
+  resolved "https://registry.yarnpkg.com/@grpc/proto-loader/-/proto-loader-0.7.13.tgz#f6a44b2b7c9f7b609f5748c6eac2d420e37670cf"
+  integrity sha512-AiXO/bfe9bmxBjxxtYxFAXGZvMaN5s8kO+jBHAJCON8rJoB5YS/D6X7ZNc6XQkuHNmyl4CYaMI1fJ/Gn27RGGw==
+  dependencies:
+    lodash.camelcase "^4.3.0"
+    long "^5.0.0"
+    protobufjs "^7.2.5"
+    yargs "^17.7.2"
+
 "@humanwhocodes/config-array@^0.11.11", "@humanwhocodes/config-array@^0.11.13":
   version "0.11.14"
   resolved "https://registry.yarnpkg.com/@humanwhocodes/config-array/-/config-array-0.11.14.tgz#d78e481a039f7566ecc9660b4ea7fe6b1fec442b"
@@ -1252,6 +1282,11 @@
     "@jridgewell/resolve-uri" "^3.1.0"
     "@jridgewell/sourcemap-codec" "^1.4.14"
 
+"@js-sdsl/ordered-map@^4.4.2":
+  version "4.4.2"
+  resolved "https://registry.yarnpkg.com/@js-sdsl/ordered-map/-/ordered-map-4.4.2.tgz#9299f82874bab9e4c7f9c48d865becbfe8d6907c"
+  integrity sha512-iUKgm52T8HOE/makSxjqoWhe95ZJA1/G1sYsGev2JDKUSS14KAgg1LHb+Ba+IPow0xflbnSkOsZcO08C7w1gYw==
+
 "@lerna/add@6.1.0":
   version "6.1.0"
   resolved "https://registry.yarnpkg.com/@lerna/add/-/add-6.1.0.tgz#0f09495c5e1af4c4f316344af34b6d1a91b15b19"
@@ -2367,6 +2402,59 @@
   resolved "https://registry.yarnpkg.com/@pinax/graph-networks-registry/-/graph-networks-registry-0.6.7.tgz#ceb994f3b31e2943b9c9d9b09dd86eb00d067c0e"
   integrity sha512-xogeCEZ50XRMxpBwE3TZjJ8RCO8Guv39gDRrrKtlpDEDEMLm0MzD3A0SQObgj7aF7qTZNRTWzsuvQdxgzw25wQ==
 
+"@protobufjs/aspromise@^1.1.1", "@protobufjs/aspromise@^1.1.2":
+  version "1.1.2"
+  resolved "https://registry.yarnpkg.com/@protobufjs/aspromise/-/aspromise-1.1.2.tgz#9b8b0cc663d669a7d8f6f5d0893a14d348f30fbf"
+  integrity sha512-j+gKExEuLmKwvz3OgROXtrJ2UG2x8Ch2YZUxahh+s1F2HZ+wAceUNLkvy6zKCPVRkU++ZWQrdxsUeQXmcg4uoQ==
+
+"@protobufjs/base64@^1.1.2":
+  version "1.1.2"
+  resolved "https://registry.yarnpkg.com/@protobufjs/base64/-/base64-1.1.2.tgz#4c85730e59b9a1f1f349047dbf24296034bb2735"
+  integrity sha512-AZkcAA5vnN/v4PDqKyMR5lx7hZttPDgClv83E//FMNhR2TMcLUhfRUBHCmSl0oi9zMgDDqRUJkSxO3wm85+XLg==
+
+"@protobufjs/codegen@^2.0.4":
+  version "2.0.4"
+  resolved "https://registry.yarnpkg.com/@protobufjs/codegen/-/codegen-2.0.4.tgz#7ef37f0d010fb028ad1ad59722e506d9262815cb"
+  integrity sha512-YyFaikqM5sH0ziFZCN3xDC7zeGaB/d0IUb9CATugHWbd1FRFwWwt4ld4OYMPWu5a3Xe01mGAULCdqhMlPl29Jg==
+
+"@protobufjs/eventemitter@^1.1.0":
+  version "1.1.0"
+  resolved "https://registry.yarnpkg.com/@protobufjs/eventemitter/-/eventemitter-1.1.0.tgz#355cbc98bafad5978f9ed095f397621f1d066b70"
+  integrity sha512-j9ednRT81vYJ9OfVuXG6ERSTdEL1xVsNgqpkxMsbIabzSo3goCjDIveeGv5d03om39ML71RdmrGNjG5SReBP/Q==
+
+"@protobufjs/fetch@^1.1.0":
+  version "1.1.0"
+  resolved "https://registry.yarnpkg.com/@protobufjs/fetch/-/fetch-1.1.0.tgz#ba99fb598614af65700c1619ff06d454b0d84c45"
+  integrity sha512-lljVXpqXebpsijW71PZaCYeIcE5on1w5DlQy5WH6GLbFryLUrBD4932W/E2BSpfRJWseIL4v/KPgBFxDOIdKpQ==
+  dependencies:
+    "@protobufjs/aspromise" "^1.1.1"
+    "@protobufjs/inquire" "^1.1.0"
+
+"@protobufjs/float@^1.0.2":
+  version "1.0.2"
+  resolved "https://registry.yarnpkg.com/@protobufjs/float/-/float-1.0.2.tgz#5e9e1abdcb73fc0a7cb8b291df78c8cbd97b87d1"
+  integrity sha512-Ddb+kVXlXst9d+R9PfTIxh1EdNkgoRe5tOX6t01f1lYWOvJnSPDBlG241QLzcyPdoNTsblLUdujGSE4RzrTZGQ==
+
+"@protobufjs/inquire@^1.1.0":
+  version "1.1.0"
+  resolved "https://registry.yarnpkg.com/@protobufjs/inquire/-/inquire-1.1.0.tgz#ff200e3e7cf2429e2dcafc1140828e8cc638f089"
+  integrity sha512-kdSefcPdruJiFMVSbn801t4vFK7KB/5gd2fYvrxhuJYg8ILrmn9SKSX2tZdV6V+ksulWqS7aXjBcRXl3wHoD9Q==
+
+"@protobufjs/path@^1.1.2":
+  version "1.1.2"
+  resolved "https://registry.yarnpkg.com/@protobufjs/path/-/path-1.1.2.tgz#6cc2b20c5c9ad6ad0dccfd21ca7673d8d7fbf68d"
+  integrity sha512-6JOcJ5Tm08dOHAbdR3GrvP+yUUfkjG5ePsHYczMFLq3ZmMkAD98cDgcT2iA1lJ9NVwFd4tH/iSSoe44YWkltEA==
+
+"@protobufjs/pool@^1.1.0":
+  version "1.1.0"
+  resolved "https://registry.yarnpkg.com/@protobufjs/pool/-/pool-1.1.0.tgz#09fd15f2d6d3abfa9b65bc366506d6ad7846ff54"
+  integrity sha512-0kELaGSIDBKvcgS4zkjz1PeddatrjYcmMWOlAuAPwAeccUrPHdUqo/J6LiymHHEiJT5NrF1UVwxY14f+fy4WQw==
+
+"@protobufjs/utf8@^1.1.0":
+  version "1.1.0"
+  resolved "https://registry.yarnpkg.com/@protobufjs/utf8/-/utf8-1.1.0.tgz#a777360b5b39a1a2e5106f8e858f2fd2d060c570"
+  integrity sha512-Vvn3zZrhQZkkBE8LSuW3em98c0FwgO4nxzv6OdSxPKJIEKY2bGbHn+mhGIPerzI4twdxaP8/0+06HBpwf345Lw==
+
 "@rushstack/node-core-library@5.12.0":
   version "5.12.0"
   resolved "https://registry.yarnpkg.com/@rushstack/node-core-library/-/node-core-library-5.12.0.tgz#de8d7d644811373ade64a926516b1ecfbe09394c"
@@ -2911,6 +2999,13 @@
   resolved "https://registry.yarnpkg.com/@types/node/-/node-20.6.1.tgz#8b589bba9b2af0128796461a0979764562687e6f"
   integrity sha512-4LcJvuXQlv4lTHnxwyHQZ3uR9Zw2j7m1C9DfuwoTFQQP4Pmu04O6IfLYgMmHoOCt0nosItLLZAH+sOrRE0Bo8g==
 
+"@types/node@>=13.7.0":
+  version "22.13.1"
+  resolved "https://registry.yarnpkg.com/@types/node/-/node-22.13.1.tgz#a2a3fefbdeb7ba6b89f40371842162fac0934f33"
+  integrity sha512-jK8uzQlrvXqEU91UxiK5J7pKHyzgnI1Qnl0QDHIgVGuolJhRb9EEl28Cj9b3rGR8B2lhFCtvIm5os8lFnO/1Ew==
+  dependencies:
+    undici-types "~6.20.0"
+
 "@types/node@^12.12.54":
   version "12.20.55"
   resolved "https://registry.yarnpkg.com/@types/node/-/node-12.20.55.tgz#c329cbd434c42164f846b909bd6f85b5537f6240"
@@ -7724,6 +7819,11 @@ log-symbols@^4.1.0:
     chalk "^4.1.0"
     is-unicode-supported "^0.1.0"
 
+long@^5.0.0:
+  version "5.2.4"
+  resolved "https://registry.yarnpkg.com/long/-/long-5.2.4.tgz#ee651d5c7c25901cfca5e67220ae9911695e99b2"
+  integrity sha512-qtzLbJE8hq7VabR3mISmVGtoXP8KGc2Z/AT8OuqlYD7JTR3oqrgwdjnk07wpj1twXxYmgDXgoKVWUG/fReSzHg==
+
 loose-envify@^1.0.0:
   version "1.4.0"
   resolved "https://registry.yarnpkg.com/loose-envify/-/loose-envify-1.4.0.tgz#71ee51fa7be4caec1a63839f7e682d8132d30caf"
@@ -9203,6 +9303,24 @@ proto-list@~1.2.1:
   resolved "https://registry.yarnpkg.com/proto-list/-/proto-list-1.2.4.tgz#212d5bfe1318306a420f6402b8e26ff39647a849"
   integrity sha512-vtK/94akxsTMhe0/cbfpR+syPuszcuwhqVjJq26CuNDgFGj682oRBXOP5MJpv2r7JtE8MsiepGIqvvOTBwn2vA==
 
+protobufjs@^7.2.5:
+  version "7.4.0"
+  resolved "https://registry.yarnpkg.com/protobufjs/-/protobufjs-7.4.0.tgz#7efe324ce9b3b61c82aae5de810d287bc08a248a"
+  integrity sha512-mRUWCc3KUU4w1jU8sGxICXH/gNS94DvI1gxqDvBzhj1JpcsimQkYiOJfwsPUykUI5ZaspFbSgmBLER8IrQ3tqw==
+  dependencies:
+    "@protobufjs/aspromise" "^1.1.2"
+    "@protobufjs/base64" "^1.1.2"
+    "@protobufjs/codegen" "^2.0.4"
+    "@protobufjs/eventemitter" "^1.1.0"
+    "@protobufjs/fetch" "^1.1.0"
+    "@protobufjs/float" "^1.0.2"
+    "@protobufjs/inquire" "^1.1.0"
+    "@protobufjs/path" "^1.1.2"
+    "@protobufjs/pool" "^1.1.0"
+    "@protobufjs/utf8" "^1.1.0"
+    "@types/node" ">=13.7.0"
+    long "^5.0.0"
+
 protocols@^2.0.0, protocols@^2.0.1:
   version "2.0.2"
   resolved "https://registry.yarnpkg.com/protocols/-/protocols-2.0.2.tgz#822e8fcdcb3df5356538b3e91bfd890b067fd0a4"
@@ -11086,7 +11204,7 @@ yargs@^16.2.0:
     y18n "^5.0.5"
     yargs-parser "^20.2.2"
 
-yargs@^17.3.1, yargs@^17.6.2:
+yargs@^17.3.1, yargs@^17.6.2, yargs@^17.7.2:
   version "17.7.2"
   resolved "https://registry.yarnpkg.com/yargs/-/yargs-17.7.2.tgz#991df39aca675a192b816e1e0363f9d75d2aa269"
   integrity sha512-7dSzzRQ++CKnNI/krKnYRV7JKKPUXMEh61soaHKg9mrWEhzFWhFnxPxGl+69cD1Ou63C13NUPCnmIcrvqCuM6w==