[CVAT] Reserve escrow funds in store_results

packages/examples/cvat/exchange-oracle/src/chain/escrow.py

@@ -121,13 +121,14 @@ def get_escrow_fund_token_decimals(chain_id: int, escrow_address: str) -> int:
     return get_token_decimals(chain_id, escrow.token)
 
 
-def get_remaining_escrow_funds(chain_id: int, escrow_address: str) -> Decimal | None:
+def get_raw_remaining_escrow_funds(chain_id: int, escrow_address: str) -> int:
     web3 = get_web3(chain_id)
-    client = EscrowClient(web3)
+    escrow_client = EscrowClient(web3)
 
-    remaining_funds = client.get_remaining_funds(escrow_address)
-    if remaining_funds is None:
-        return None
+    return escrow_client.get_remaining_funds(escrow_address)
 
+
+def get_remaining_escrow_funds(chain_id: int, escrow_address: str) -> Decimal:
+    remaining_funds = get_raw_remaining_escrow_funds(chain_id, escrow_address)
     token_decimals = get_escrow_fund_token_decimals(chain_id, escrow_address)
     return Decimal(remaining_funds) / Decimal(10**token_decimals)

packages/examples/cvat/exchange-oracle/src/core/annotation_meta.py

@@ -11,6 +11,8 @@ class JobMeta(BaseModel):
     assignment_id: str
     start_frame: int
     stop_frame: int
+    assignment_bounty: str | None = None
+    "Assignment reward, a decimal value in the escrow fund token units"
 
 
 class AnnotationMeta(BaseModel):

packages/examples/cvat/exchange-oracle/src/handlers/job_creation/utils.py

@@ -57,9 +57,6 @@ def _get_assignment_bounty_from_escrow(
         return None
 
     funds = get_remaining_escrow_funds(chain_id, escrow_address)
-    if funds is None:
-        return None
-
     return str(funds / job_count)
 
 

packages/examples/cvat/exchange-oracle/src/handlers/job_export/results.py

@@ -43,6 +43,7 @@ def prepare_annotation_metafile(jobs: list[Job]) -> FileDescriptor:
                 task_id=job.cvat_task_id,
                 start_frame=job.start_frame,
                 stop_frame=job.stop_frame,
+                assignment_bounty=job.project.assignment_bounty,
             )
             for job in jobs
         ]

packages/examples/cvat/exchange-oracle/tests/assets/utils/gen_audio_validation_fixture.py

@@ -25,6 +25,7 @@
 from __future__ import annotations
 
 import tempfile
+from decimal import Decimal
 from pathlib import Path
 
 INPUT_BUCKET = "datasets"
@@ -115,6 +116,10 @@ def upload_input(tmp_dir: Path):
         with (
             patch.object(handlers, "get_escrow_manifest", return_value=manifest),
             patch("src.handlers.job_creation.builders.audio.transcription.cvat_api", cvat_api),
+            patch(
+                "src.handlers.job_creation.utils.get_remaining_escrow_funds",
+                return_value=Decimal(10),
+            ),
         ):
             handlers.create_task(ESCROW_ADDRESS, chain_id)
 

packages/examples/cvat/recording-oracle/alembic/versions/b7c1f0a4d2e3_add_assignment_bounty.py

@@ -0,0 +1,26 @@
+"""
+Add assignment bounty
+
+Revision ID: b7c1f0a4d2e3
+Revises: a5907f01ac2d
+Create Date: 2026-07-29 14:30:00.000000
+
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "b7c1f0a4d2e3"
+down_revision = "a5907f01ac2d"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("validation_results", sa.Column("assignment_bounty", sa.String(), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("validation_results", "assignment_bounty")

packages/examples/cvat/recording-oracle/src/chain/escrow.py

@@ -1,12 +1,13 @@
 import json
+from decimal import Decimal
 
 import httpx2
 from human_protocol_sdk.constants import ChainId, Status
 from human_protocol_sdk.encryption import Encryption, EncryptionUtils
 from human_protocol_sdk.escrow import EscrowClient, EscrowData, EscrowUtils
 from human_protocol_sdk.utils import validate_url
 
-from src.chain.web3 import get_web3
+from src.chain.web3 import get_token_decimals, get_web3
 from src.core.config import Config
 from src.core.types import OracleWebhookTypes
 
@@ -77,11 +78,36 @@ def get_escrow_manifest(chain_id: int, escrow_address: str) -> dict:
     return json.loads(manifest_content)
 
 
-def store_results(chain_id: int, escrow_address: str, url: str, hash: str) -> None:
+def get_escrow_fund_token_decimals(chain_id: int, escrow_address: str) -> int:
+    """
+    ERC-20 decimals: divide the raw token amount by 10**decimals for the user representation
+    (https://eips.ethereum.org/EIPS/eip-20).
+    """
+
+    escrow = get_escrow(chain_id, escrow_address)
+    return get_token_decimals(chain_id, escrow.token)
+
+
+def get_raw_remaining_escrow_funds(chain_id: int, escrow_address: str) -> int:
+    web3 = get_web3(chain_id)
+    escrow_client = EscrowClient(web3)
+
+    return escrow_client.get_remaining_funds(escrow_address)
+
+
+def get_remaining_escrow_funds(chain_id: int, escrow_address: str) -> Decimal:
+    remaining_funds = get_raw_remaining_escrow_funds(chain_id, escrow_address)
+    token_decimals = get_escrow_fund_token_decimals(chain_id, escrow_address)
+    return Decimal(remaining_funds) / Decimal(10**token_decimals)
+
+
+def store_results(
+    chain_id: int, escrow_address: str, url: str, hash: str, funds_to_reserve: int | None = None
+) -> None:
     web3 = get_web3(chain_id)
     escrow_client = EscrowClient(web3)
 
-    escrow_client.store_results(escrow_address, url, hash)
+    escrow_client.store_results(escrow_address, url, hash, funds_to_reserve=funds_to_reserve)
 
 
 def get_available_webhook_types(

packages/examples/cvat/recording-oracle/src/chain/web3.py

@@ -9,6 +9,16 @@
 from src.core.config import Config
 from src.core.types import Networks
 
+decimals_abi = [
+    {
+        "constant": True,
+        "inputs": [],
+        "name": "decimals",
+        "outputs": [{"name": "", "type": "uint8"}],
+        "type": "function",
+    }
+]  # ABI for fetching token decimals (ERC-20 optional method; https://eips.ethereum.org/EIPS/eip-20)
+
 
 def get_web3(chain_id: Networks):
     match chain_id:
@@ -81,3 +91,9 @@ def validate_address(escrow_address: str) -> str:
     if not Web3.is_address(escrow_address):
         raise ValueError(f"{escrow_address} is not a correct Web3 address")
     return Web3.to_checksum_address(escrow_address)
+
+
+def get_token_decimals(chain_id: int, token_address: str) -> int:
+    w3 = get_web3(chain_id)
+    contract = w3.eth.contract(address=w3.to_checksum_address(token_address), abi=decimals_abi)
+    return contract.functions.decimals().call()

packages/examples/cvat/recording-oracle/src/core/annotation_meta.py

@@ -18,6 +18,8 @@ class JobMeta(BaseModel):
     assignment_id: str
     start_frame: int
     stop_frame: int
+    assignment_bounty: str | None = None
+    "Assignment reward, a decimal value in the escrow fund token units"
 
     @property
     def job_frame_range(self) -> Iterator[int]:

packages/examples/cvat/recording-oracle/src/entrypoints/debug.py

@@ -79,10 +79,11 @@ def patched_store_results(
         escrow_address,
         url,
         hash,
+        funds_to_reserve=None,
     ) -> None:
         logger.info(
             f"DEV: Would store results for escrow '{escrow_address}@{chain_id}' "
-            f"on chain: {url}, {hash}"
+            f"on chain: {url}, {hash}, reserving {funds_to_reserve}"
         )
 
     with mock.patch("src.chain.escrow.store_results", patched_store_results):

packages/examples/cvat/recording-oracle/src/handlers/completion/handlers.py

@@ -1,11 +1,13 @@
 import io
 import os
+from decimal import Decimal
 from logging import Logger
 
 from sqlalchemy.orm import Session
 
 import src.core.annotation_meta as annotation
 import src.core.validation_meta as validation
+import src.services.validation as db_service
 import src.services.webhook as oracle_db_service
 from src.chain import escrow
 from src.core.config import Config
@@ -80,6 +82,66 @@ def export(self):
     def _compose_validation_results_bucket_filename(self, filename: str) -> str:
         return f"{self.escrow_address}@{self.chain_id}/{filename}"
 
+    def _compute_total_bounty(self) -> Decimal | None:
+        """
+        Computes the total reward for the final assignments.
+        Returns None, if the reward is unknown for the assignments.
+        """
+
+        assert self.annotation_meta is not None
+
+        assignment_bounties = {
+            job_meta.assignment_id: db_service.get_validation_result_by_assignment_id(
+                self.db_session, job_meta.assignment_id
+            ).assignment_bounty
+            for job_meta in self.annotation_meta.jobs
+        }
+
+        assignments_without_bounty = [
+            assignment_id for assignment_id, bounty in assignment_bounties.items() if bounty is None
+        ]
+        if assignments_without_bounty:
+            # The assignment rewards are expected to be either known for all the assignments
+            # or unknown for all of them, otherwise the total reward can't be computed
+            if len(assignments_without_bounty) != len(assignment_bounties):
+                raise Exception(
+                    f"Result uploading for escrow_address={self.escrow_address}: "
+                    f"{len(assignments_without_bounty)} of {len(assignment_bounties)} assignments "
+                    "have no reward specified. "
+                    "Either all the assignments must have bounty specified or none of them."
+                )
+
+            return None
+
+        return sum((Decimal(bounty) for bounty in assignment_bounties.values()), start=Decimal(0))
+
+    def _compute_funds_to_reserve(self, total_bounty: Decimal | None) -> int:
+        """
+        Returns the escrow funds to be reserved for the payouts, in the raw token units.
+        All the remaining funds are reserved, the requested reward is only validated.
+        """
+
+        remaining_funds = escrow.get_raw_remaining_escrow_funds(self.chain_id, self.escrow_address)
+
+        if total_bounty is not None:
+            token_decimals = escrow.get_escrow_fund_token_decimals(
+                self.chain_id, self.escrow_address
+            )
+            requested_funds = total_bounty * 10**token_decimals
+            if requested_funds > remaining_funds:
+                raise Exception(
+                    f"Result uploading for escrow_address={self.escrow_address}: "
+                    f"the total assignment reward ({requested_funds}) exceeds "
+                    f"the remaining escrow funds ({remaining_funds})"
+                )
+
+        self.logger.info(
+            f"Result uploading for escrow_address={self.escrow_address}: "
+            f"will reserve {remaining_funds} funds on the escrow."
+        )
+
+        return remaining_funds
+
     def _handle_result(self, export_result: FinalResult):
         logger = self.logger
         escrow_address = self.escrow_address
@@ -112,11 +174,14 @@ def _handle_result(self, export_result: FinalResult):
             validation_metafile,
         )
 
+        funds_to_reserve = self._compute_funds_to_reserve(self._compute_total_bounty())
+
         escrow.store_results(
             chain_id,
             escrow_address,
             Config.storage_config.bucket_url() + os.path.dirname(recor_merged_annotations_path),  # noqa: PTH120
             compute_resulting_annotations_hash(export_result.resulting_annotations),
+            funds_to_reserve=funds_to_reserve,
         )
 
         oracle_db_service.outbox.create_webhook(

packages/examples/cvat/recording-oracle/src/handlers/validation/intermediate_results.py

@@ -473,6 +473,7 @@ def process_intermediate_results(  # noqa: PLR0912
                 annotator_wallet_address=job_meta.annotator_wallet_address,
                 annotation_quality=job_results[job_meta.job_id],
                 assignment_id=job_meta.assignment_id,
+                assignment_bounty=job_meta.assignment_bounty,
             )
         else:
             assignment_validation_result_id = assignment_validation_result.id

packages/examples/cvat/recording-oracle/src/models/validation.py

@@ -53,6 +53,8 @@ class ValidationResult(Base):
     assignment_id = Column(String, unique=True, nullable=False)
     annotator_wallet_address = Column(String, nullable=False)
     annotation_quality = Column(Float, nullable=False)
+    assignment_bounty = Column(String, nullable=True)
+    "Assignment reward, a decimal value in the escrow fund token units"
 
     job: Mapped[Job] = relationship(back_populates="validation_results")
 

packages/examples/cvat/recording-oracle/src/services/validation.py

@@ -98,6 +98,7 @@ def create_validation_result(
     annotator_wallet_address: str,
     annotation_quality: float,
     assignment_id: str,
+    assignment_bounty: str | None = None,
 ) -> str:
     obj_id = str(uuid.uuid4())
     obj = ValidationResult(
@@ -106,6 +107,7 @@ def create_validation_result(
         annotator_wallet_address=annotator_wallet_address,
         annotation_quality=annotation_quality,
         assignment_id=assignment_id,
+        assignment_bounty=assignment_bounty,
     )
 
     session.add(obj)

packages/examples/cvat/recording-oracle/tests/integration/chain/test_escrow.py

@@ -130,7 +130,26 @@ def test_store_results(self):
                 store_results(chain_id, escrow_address, DEFAULT_MANIFEST_URL, DEFAULT_HASH) is None
             )
             mock_client_cls.return_value.store_results.assert_called_once_with(
-                escrow_address, DEFAULT_MANIFEST_URL, DEFAULT_HASH
+                escrow_address, DEFAULT_MANIFEST_URL, DEFAULT_HASH, funds_to_reserve=None
+            )
+
+    def test_store_results_with_funds_to_reserve(self):
+        with (
+            patch("src.chain.escrow.get_web3"),
+            patch("src.chain.escrow.EscrowClient") as mock_client_cls,
+        ):
+            assert (
+                store_results(
+                    chain_id,
+                    escrow_address,
+                    DEFAULT_MANIFEST_URL,
+                    DEFAULT_HASH,
+                    funds_to_reserve=42,
+                )
+                is None
+            )
+            mock_client_cls.return_value.store_results.assert_called_once_with(
+                escrow_address, DEFAULT_MANIFEST_URL, DEFAULT_HASH, funds_to_reserve=42
             )
 
     def test_store_results_invalid_url(self):