Adding gold units and functionality

.github/workflows/cypress-end-to-end-tests.yml

@@ -69,6 +69,8 @@ jobs:
 
       - name: 🐍 Installing python
         uses: actions/setup-python@v2
+        with:
+          python-version: 3.8
 
       - name: 🪨 Setup Node
         uses: actions/setup-node@v3
@@ -114,6 +116,8 @@ jobs:
 
       - name: 🐍 Installing python
         uses: actions/setup-python@v2
+        with:
+          python-version: 3.8
 
       - name: 🪨 Setup Node
         uses: actions/setup-node@v3
@@ -159,6 +163,8 @@ jobs:
 
       - name: 🐍 Installing python
         uses: actions/setup-python@v2
+        with:
+          python-version: 3.8
 
       - name: 🪨 Setup Node
         uses: actions/setup-node@v3
@@ -203,6 +209,8 @@ jobs:
 
       - name: 🐍 Installing python
         uses: actions/setup-python@v2
+        with:
+          python-version: 3.8
 
       - name: 🪨 Setup Node
         uses: actions/setup-node@v3
@@ -249,6 +257,8 @@ jobs:
 
       - name: 🐍 Installing python
         uses: actions/setup-python@v2
+        with:
+          python-version: 3.8
 
       - name: 🪨 Setup Node
         uses: actions/setup-node@v3
@@ -293,6 +303,8 @@ jobs:
         uses: actions/checkout@v2
       - name: 🐍 Installing python
         uses: actions/setup-python@v2
+        with:
+          python-version: 3.8
 
       - name: 🪨 Setup Node
         uses: actions/setup-node@v3

.github/workflows/deploy-docs.yml

@@ -21,6 +21,8 @@ jobs:
           node-version: 16.16.0
           cache: yarn
       - uses: actions/setup-python@v2
+        with:
+          python-version: 3.8
       - name: 🤖 Install Mephisto
         run: |
           cd ../../

.github/workflows/npm-check.yml

@@ -10,6 +10,8 @@ jobs:
     steps:
       - uses: actions/checkout@v2
       - uses: actions/setup-python@v2
+        with:
+          python-version: 3.8
       - name: Install Mephisto
         run: pip install -e .
       - name: Run version sync script

.github/workflows/pre-commit.yml

@@ -11,4 +11,6 @@ jobs:
     steps:
       - uses: actions/checkout@v2
       - uses: actions/setup-python@v2
+        with:
+          python-version: 3.8
       - uses: pre-commit/[email protected]

.github/workflows/test-deploy-docs.yml

@@ -23,6 +23,8 @@ jobs:
           node-version: 16.16.0
           cache: yarn
       - uses: actions/setup-python@v2
+        with:
+          python-version: 3.8
 
       - name: 🤖 Install Mephisto
         run: |

.github/workflows/version-sync.yml

@@ -11,6 +11,8 @@ jobs:
     steps:
       - uses: actions/checkout@v2
       - uses: actions/setup-python@v2
+        with:
+          python-version: 3.8
       - name: Install Mephisto
         run: pip install -e .
       - name: Run version sync script

mephisto/abstractions/blueprints/abstract/static_task/static_blueprint.py

@@ -21,6 +21,11 @@
     ScreenTaskRequiredArgs,
     ScreenTaskSharedState,
 )
+from mephisto.abstractions.blueprints.mixins.use_gold_unit import (
+    UseGoldUnit,
+    UseGoldUnitArgs,
+    GoldUnitSharedState,
+)
 from mephisto.data_model.assignment import InitializationData
 from mephisto.abstractions.blueprints.abstract.static_task.static_agent_state import (
     StaticAgentState,
@@ -53,7 +58,7 @@
 
 @dataclass
 class SharedStaticTaskState(
-    ScreenTaskSharedState, OnboardingSharedState, SharedTaskState
+    ScreenTaskSharedState, OnboardingSharedState, GoldUnitSharedState, SharedTaskState
 ):
     static_task_data: Iterable[Any] = field(
         default_factory=list,
@@ -71,7 +76,7 @@ class SharedStaticTaskState(
 
 @dataclass
 class StaticBlueprintArgs(
-    ScreenTaskRequiredArgs, OnboardingRequiredArgs, BlueprintArgs
+    ScreenTaskRequiredArgs, OnboardingRequiredArgs, UseGoldUnitArgs, BlueprintArgs
 ):
     _blueprint_type: str = BLUEPRINT_TYPE_STATIC
     _group: str = field(
@@ -107,7 +112,7 @@ class StaticBlueprintArgs(
     )
 
 
-class StaticBlueprint(ScreenTaskRequired, OnboardingRequired, Blueprint):
+class StaticBlueprint(ScreenTaskRequired, OnboardingRequired, UseGoldUnit, Blueprint):
     """
     Abstract blueprint for a task that runs without any extensive backend.
     These are generally one-off tasks sending data to the frontend and then

mephisto/abstractions/blueprints/mixins/screen_task_required.py

@@ -118,7 +118,9 @@ def init_screening_config(
         self.use_screening_task = args.blueprint.get("use_screening_task", False)
         if not self.use_screening_task:
             return
-
+        assert (
+            args.blueprint.get("use_golds") is not True
+        ), "Gold units currently cannot be used with screening units"
         # Runs that are using a qualification task should be able to assign
         # a specially generated unit to unqualified workers
         self.passed_qualification_name = args.blueprint.passed_qualification_name

mephisto/abstractions/blueprints/mixins/use_gold_unit.py

@@ -53,6 +53,16 @@ class UseGoldUnitArgs:
             "help": ("Basename for a qualification that tracks gold completion rates")
         },
     )
+    max_gold_units: int = field(
+        default=MISSING,
+        metadata={
+            "help": (
+                "The maximum number of gold units that can be launched "
+                "with this batch, specified to limit the number of golds "
+                "you may need to pay out for."
+            )
+        },
+    )
     use_golds: bool = field(
         default=False,
         metadata={"help": ("Whether or not to use gold tasks in this run.")},
@@ -198,6 +208,8 @@ def init_gold_config(
 
         self.min_golds = args.blueprint.min_golds
         self.max_incorrect_golds = args.blueprint.max_incorrect_golds
+        self.gold_units_launched = 0
+        self.gold_unit_cap = args.blueprint.max_gold_units
 
         find_or_create_qualification(task_run.db, self.golds_correct_qual_name)
         find_or_create_qualification(task_run.db, self.golds_failed_qual_name)
@@ -213,6 +225,15 @@ def assert_mixin_args(cls, args: "DictConfig", shared_state: "SharedTaskState"):
             "Can only run this task type with one allowed concurrent unit at a time per worker, to ensure "
             "golds are completed in order."
         )
+        assert (
+            args.blueprint.get("use_screening_task") is not True
+        ), "Gold units currently cannot be used with screening units"
+        max_gold_units = args.blueprint.max_gold_units
+        assert max_gold_units is not None, (
+            "You must supply a blueprint.max_gold_units argument to set the maximum number of "
+            "additional units you will pay out for evaluating on gold units. Note that you "
+            "do pay for gold units, they are just like any other units."
+        )
         gold_qualification_base = args.blueprint.gold_qualification_base
         assert (
             gold_qualification_base is not None
@@ -257,6 +278,9 @@ def should_produce_gold_for_worker(self, worker: "Worker") -> bool:
             completed_units, correct_units, incorrect_units, self.max_incorrect_golds
         ):
             return False
+        if correct_units >= self.min_golds:
+            if self.gold_units_launched >= self.gold_unit_cap:
+                return False  # they qualify, but we don't have golds to launch
         return self.worker_needs_gold(
             completed_units, correct_units, incorrect_units, self.min_golds
         )
@@ -278,7 +302,10 @@ def update_qualified_status(self, worker: "Worker") -> bool:
     def get_gold_unit_data_for_worker(
         self, worker: "Worker"
     ) -> Optional[Dict[str, Any]]:
+        if self.gold_units_launched >= self.gold_unit_cap:
+            return None
         try:
+            self.gold_units_launched += 1
             return self.get_gold_for_worker(worker)
         except Exception as e:
             logger.warning(f"Could not generate gold for {worker} due to {e}")

mephisto/abstractions/blueprints/remote_procedure/remote_procedure_blueprint.py

@@ -9,17 +9,22 @@
     BlueprintArgs,
     SharedTaskState,
 )
+from dataclasses import dataclass, field
 from mephisto.abstractions.blueprints.mixins.onboarding_required import (
     OnboardingRequired,
     OnboardingSharedState,
     OnboardingRequiredArgs,
 )
-from dataclasses import dataclass, field
 from mephisto.abstractions.blueprints.mixins.screen_task_required import (
     ScreenTaskRequired,
     ScreenTaskRequiredArgs,
     ScreenTaskSharedState,
 )
+from mephisto.abstractions.blueprints.mixins.use_gold_unit import (
+    UseGoldUnit,
+    UseGoldUnitArgs,
+    GoldUnitSharedState,
+)
 from mephisto.data_model.assignment import InitializationData
 from mephisto.abstractions.blueprints.remote_procedure.remote_procedure_agent_state import (
     RemoteProcedureAgentState,
@@ -59,7 +64,7 @@
 
 @dataclass
 class SharedRemoteProcedureTaskState(
-    ScreenTaskSharedState, OnboardingSharedState, SharedTaskState
+    ScreenTaskSharedState, OnboardingSharedState, GoldUnitSharedState, SharedTaskState
 ):
     function_registry: Optional[
         Mapping[
@@ -75,7 +80,7 @@ class SharedRemoteProcedureTaskState(
 
 @dataclass
 class RemoteProcedureBlueprintArgs(
-    ScreenTaskRequiredArgs, OnboardingRequiredArgs, BlueprintArgs
+    ScreenTaskRequiredArgs, OnboardingRequiredArgs, UseGoldUnitArgs, BlueprintArgs
 ):
     _blueprint_type: str = BLUEPRINT_TYPE_REMOTE_PROCEDURE
     _group: str = field(
@@ -114,7 +119,9 @@ class RemoteProcedureBlueprintArgs(
 
 
 @register_mephisto_abstraction()
-class RemoteProcedureBlueprint(ScreenTaskRequired, OnboardingRequired, Blueprint):
+class RemoteProcedureBlueprint(
+    ScreenTaskRequired, OnboardingRequired, UseGoldUnit, Blueprint
+):
     """Blueprint for a task that runs a parlai chat"""
 
     AgentStateClass: ClassVar[Type["AgentState"]] = RemoteProcedureAgentState

mephisto/operations/client_io_handler.py

@@ -371,6 +371,7 @@ def enqueue_agent_details(self, request_id: str, additional_data: Dict[str, Any]
         )
         self.process_outgoing_queue(self.message_queue)
         self.log_metrics_for_packet(self.request_id_to_packet[request_id])
+        # TODO Sometimes this request ID is lost, and we don't quite know why
         del self.request_id_to_channel_id[request_id]
         del self.request_id_to_packet[request_id]