diff --git a/compute_worker/compute_worker.py b/compute_worker/compute_worker.py index 243dcd16d..84fcf1825 100644 --- a/compute_worker/compute_worker.py +++ b/compute_worker/compute_worker.py @@ -112,6 +112,9 @@ def to_bool(val): CODALAB_IGNORE_CLEANUP_STEP = to_bool(get("CODALAB_IGNORE_CLEANUP_STEP")) WORKER_BUNDLE_URL_REWRITE = get("WORKER_BUNDLE_URL_REWRITE", "").strip() + HUMAN_IN_THE_LOOP = ( + get("HUMAN_IN_THE_LOOP", "false").lower() == "true" + ) # ----------------------------------------------- @@ -131,6 +134,7 @@ class SubmissionStatus: SUBMITTED = "Submitted" PREPARING = "Preparing" RUNNING = "Running" + AWAITING_VALIDATION = "Awaiting validation" SCORING = "Scoring" FINISHED = "Finished" FAILED = "Failed" @@ -141,6 +145,7 @@ class SubmissionStatus: SUBMITTED, PREPARING, RUNNING, + AWAITING_VALIDATION, SCORING, FINISHED, FAILED, @@ -308,14 +313,35 @@ def rewrite_bundle_url_if_needed(url): def run_wrapper(run_args): # We need to convert the UUID given by celery into a byte like object otherwise things will break run_args.update(secret=str(run_args["secret"])) + logger.info(f"Received run arguments: \n {colorize_run_args(json.dumps(run_args))}") + logger.info( + "HITL configuration : " + f"task={run_args.get('human_in_the_loop', False)} " + f"compute_worker={Settings.HUMAN_IN_THE_LOOP}" + ) + run = Run(run_args) + try: run.prepare() + run.validate_hitl_configuration() run.start() + if run.is_scoring: - run.push_scores() - run.push_output() + if run.human_in_the_loop: + run._update_status(SubmissionStatus.AWAITING_VALIDATION) + run.wait_for_human_validation() + run.send_final_detailed_results() + run.push_scores() + run.push_output() + else: + run.push_scores() + run.push_output() + else: + run.push_output() + + run._update_status(SubmissionStatus.FINISHED) except DockerImagePullException as e: msg = str(e).strip() if msg: @@ -470,6 +496,8 @@ def __init__(self, run_args): self.prediction_result = run_args["prediction_result"] self.scoring_result = run_args.get("scoring_result") self.execution_time_limit = run_args["execution_time_limit"] + # ----- HITL ------ + self.human_in_the_loop = run_args.get("human_in_the_loop", False) # stdout and stderr self.stdout, self.stderr, self.ingestion_stdout, self.ingestion_stderr = ( self._get_stdout_stderr_file_names(run_args) @@ -513,6 +541,13 @@ def __init__(self, run_args): async def watch_detailed_results(self): """Watches files alongside scoring + program containers, currently only used for detailed_results.html""" + + if self.human_in_the_loop: + logger.info( + "HITL enabled: skipping detailed_results streaming" + ) + return + if not self.detailed_results_url: return file_path = self.get_detailed_results_file_path() @@ -596,6 +631,56 @@ async def send_detailed_results(self, file_path): logger.exception(e) return + finally: + if websocket is not None: + try: + await websocket.close() + except Exception as e: + logger.exception(e) + + def send_final_detailed_results(self): + if not self.detailed_results_url: + return + + file_path = self.get_detailed_results_file_path() + + if not file_path: + logger.info("No detailed_results.html found") + return + + logger.info( + f"Uploading final detailed results {file_path} - {self.detailed_results_url}" + ) + + self._put_file( + self.detailed_results_url, + file=file_path, + content_type="text/html", + ) + + websocket_url = f"{self.websocket_url}?kind=detailed_results" + + try: + websocket = asyncio.run( + asyncio.wait_for( + websockets.connect(websocket_url), + timeout=30.0, + ) + ) + + asyncio.run( + websocket.send( + json.dumps( + { + "kind": "detailed_result_update", + } + ) + ) + ) + + except Exception as e: + logger.exception(e) + def _get_stdout_stderr_file_names(self, run_args): # run_args should be the run_args argument passed to __init__ from the run_wrapper. if not self.is_scoring: @@ -1282,6 +1367,15 @@ def prepare(self): self._get_container_image(self.container_image) self._update_status(SubmissionStatus.RUNNING) + def validate_hitl_configuration(self): + if self.human_in_the_loop != Settings.HUMAN_IN_THE_LOOP: + raise SubmissionException( + "Task rejected because the Site Worker and Compute Worker " + "do not have the same HUMAN_IN_THE_LOOP configuration " + f"(task={self.human_in_the_loop}, " + f"compute_worker={Settings.HUMAN_IN_THE_LOOP})." + ) + def start(self): logger.info(f"Preparing to run: {ProgramKind.SCORING_PROGRAM if self.is_scoring else ProgramKind.INGESTION_PROGRAM}") @@ -1308,9 +1402,10 @@ def start(self): ) # During scoring we watch for detailed results - tasks.append( - self.watch_detailed_results() - ) + if not self.human_in_the_loop: + tasks.append( + self.watch_detailed_results() + ) else: # During ingestion we run ingestion program directory and submission directory tasks.extend([ @@ -1429,9 +1524,15 @@ def start(self): # Check if scoring program failed # We have can have 2 or 3 gathered tasks: 3 gathered tasks in case when `ingestion_only_during_scoring` is True, 2 otherwise if self.ingestion_only_during_scoring: - program_results, _, _ = task_results + if self.human_in_the_loop: + program_results, _ = task_results + else: + program_results, _, _ = task_results else: - program_results, _ = task_results + if self.human_in_the_loop: + (program_results,) = task_results + else: + program_results, _ = task_results # Gather returns either normal values or exception instances when return_exceptions=True had_async_exc = isinstance( program_results, BaseException @@ -1445,11 +1546,59 @@ def start(self): ) # Raise so upstream marks failed immediately raise SubmissionException("Child task failed or non-zero return code") - self._update_status(SubmissionStatus.FINISHED) + + if not self.human_in_the_loop: + self._update_status(SubmissionStatus.FINISHED) else: self._update_status(SubmissionStatus.SCORING) + def wait_for_human_validation(self): + container_output_dir = self.output_dir + host_output_dir = self._get_host_path(self.output_dir) + + scores_path = os.path.join(host_output_dir, "scores.json") + if not os.path.exists(os.path.join(container_output_dir, "scores.json")): + scores_path = os.path.join(host_output_dir, "scores.txt") + + approved_container = os.path.join(container_output_dir, "hitl_approved") + rejected_container = os.path.join(container_output_dir, "hitl_rejected") + approved_host = os.path.join(host_output_dir, "hitl_approved") + rejected_host = os.path.join(host_output_dir, "hitl_rejected") + + logger.info("=" * 60) + logger.info(f"HUMAN IN THE LOOP — submission {self.submission_id}") + logger.info("Inspect the scores file:") + logger.info(f" cat {scores_path}") + logger.info(f"To approve : touch {approved_host}") + logger.info(f"To reject : touch {rejected_host}") + logger.info("=" * 60) + + poll_interval = 3 + max_wait = 60 * 60 * 24 + + elapsed = 0 + + logger.info( + "Waiting for human validation..." + ) + while elapsed < max_wait: + if os.path.exists(approved_container): + logger.info(f"HITL: submission {self.submission_id} approved, sending scores.") + return + if os.path.exists(rejected_container): + raise SubmissionException( + f"HITL: scores rejected by the compute node operator " + f"(submission {self.submission_id})" + ) + time.sleep(poll_interval) + elapsed += poll_interval + + raise SubmissionException( + f"HITL: 24h timeout reached without validation " + f"(submission {self.submission_id})" + ) + def push_scores(self): """This is only ran at the end of the scoring step""" # POST to some endpoint: diff --git a/src/apps/api/serializers/competitions.py b/src/apps/api/serializers/competitions.py index a0fb14fda..3a566446e 100644 --- a/src/apps/api/serializers/competitions.py +++ b/src/apps/api/serializers/competitions.py @@ -272,7 +272,8 @@ class Meta: 'contact_email', 'report', 'whitelist_emails', - 'forum_enabled' + 'forum_enabled', + 'enable_human_in_the_loop' ) def validate_phases(self, phases): @@ -417,7 +418,8 @@ class Meta: 'contact_email', 'report', 'whitelist_emails', - 'forum_enabled' + 'forum_enabled', + 'enable_human_in_the_loop' ) def get_leaderboards(self, instance): diff --git a/src/apps/competitions/models.py b/src/apps/competitions/models.py index f919ae432..97d7ff057 100644 --- a/src/apps/competitions/models.py +++ b/src/apps/competitions/models.py @@ -89,6 +89,7 @@ class Competition(models.Model): # If true, forum is enabled (default=True) forum_enabled = models.BooleanField(default=True) + enable_human_in_the_loop = models.BooleanField(default=False) def __str__(self): return f"competition-{self.title}-{self.pk}-{self.competition_type}" @@ -437,6 +438,7 @@ class Submission(models.Model): PREPARING = "Preparing" RUNNING = "Running" SCORING = "Scoring" + AWAITING_VALIDATION = "Awaiting validation" CANCELLED = "Cancelled" FINISHED = "Finished" FAILED = "Failed" @@ -448,6 +450,7 @@ class Submission(models.Model): (PREPARING, "Preparing"), (RUNNING, "Running"), (SCORING, "Scoring"), + (AWAITING_VALIDATION, "Awaiting validation"), (CANCELLED, "Cancelled"), (FINISHED, "Finished"), (FAILED, "Failed"), diff --git a/src/apps/competitions/tasks.py b/src/apps/competitions/tasks.py index 3d7d8abcb..35bbebd32 100644 --- a/src/apps/competitions/tasks.py +++ b/src/apps/competitions/tasks.py @@ -58,6 +58,7 @@ "contact_email", "fact_sheet", "forum_enabled", + "enable_human_in_the_loop" ] TASK_FIELDS = [ @@ -156,6 +157,11 @@ def _get_user_group_queues(user, competition): def _send_to_compute_worker(submission, is_scoring): + hitl_active = ( + submission.phase.competition.enable_human_in_the_loop + and submission.queue is not None + ) + run_args = { "user_pk": submission.owner.pk, "submissions_api_url": settings.SUBMISSIONS_API_URL, @@ -166,6 +172,7 @@ def _send_to_compute_worker(submission, is_scoring): ), "id": submission.pk, "is_scoring": is_scoring, + "human_in_the_loop": hitl_active, } if ( @@ -249,7 +256,7 @@ def _send_to_compute_worker(submission, is_scoring): logger.info(run_args) # Pad timelimit so worker has time to cleanup - time_padding = 60 * 20 # 20 minutes + time_padding = 60 * 20 time_limit = submission.phase.execution_time_limit + time_padding effective_queue = submission.queue or submission.phase.competition.queue @@ -260,6 +267,17 @@ def _send_to_compute_worker(submission, is_scoring): submission.queue = effective_queue submission.save(update_fields=["queue"]) + if is_scoring and hitl_active: + time_limit = 60 * 60 * 25 + + if ( + submission.phase.competition.queue + ): # if the competition is running on a custom queue, not the default queue + submission.queue = submission.phase.competition.queue + run_args["execution_time_limit"] = ( + submission.phase.execution_time_limit + ) # use the competition time limit + submission.save(update_fields=["queue"]) if submission.status == Submission.SUBMITTING: submission.status = Submission.SUBMITTED submission.save(update_fields=["status"]) @@ -357,6 +375,23 @@ def _run_submission(submission_pk, task_pks=None, is_scoring=False): ) submission = qs.get(pk=submission_pk) + # ── HUMAN IN THE LOOP SECURE (private CW only) + if is_scoring and submission.phase.competition.enable_human_in_the_loop: + if submission.queue is None: + logger.error( + f"HITL: submission {submission_pk} rejected — " + f"Human in the Loop is enabled but no private queue is configured " + f"(competition or participant group)." + ) + submission.status = Submission.FAILED + submission.status_details = ( + "This competition requires Human in the Loop validation, but no " + "private compute queue is configured. Contact the organizer." + ) + submission.save(update_fields=["status", "status_details"]) + return + # ── HITL SECURE (private CW only) + if submission.is_specific_task_re_run: # Should only be one task for a specified task submission tasks = Task.objects.filter(pk__in=task_pks) diff --git a/src/static/riot/competitions/detail/submission_manager.tag b/src/static/riot/competitions/detail/submission_manager.tag index 3f41a69eb..37530345c 100644 --- a/src/static/riot/competitions/detail/submission_manager.tag +++ b/src/static/riot/competitions/detail/submission_manager.tag @@ -107,7 +107,8 @@ { submission.status } - + + + +
+ +
+ + +
+ + + + + + +
+
@@ -314,6 +335,12 @@ self.data["auto_run_submissions"] = self.refs.auto_run_submissions.checked self.data["can_participants_make_submissions_public"] = self.refs.can_participants_make_submissions_public.checked self.data["forum_enabled"] = self.refs.forum_enabled.checked + self.data["enable_human_in_the_loop"] = self.refs.enable_human_in_the_loop.checked + const hitlChecked = self.refs.enable_human_in_the_loop.checked + const hasPrivateQueue = !!(self.data["queue"] && self.data["queue"] !== "") + if (self.refs.hitl_warning) { + self.refs.hitl_warning.style.display = (hitlChecked && !hasPrivateQueue) ? 'block' : 'none' + } self.data["make_programs_available"] = self.refs.make_programs_available.checked self.data["make_input_data_available"] = self.refs.make_input_data_available.checked self.data["docker_image"] = $(self.refs.docker_image).val() @@ -452,6 +479,7 @@ self.refs.auto_run_submissions.checked = competition.auto_run_submissions self.refs.can_participants_make_submissions_public.checked = competition.can_participants_make_submissions_public self.refs.forum_enabled.checked = competition.forum_enabled + self.refs.enable_human_in_the_loop.checked = competition.enable_human_in_the_loop self.refs.make_programs_available.checked = competition.make_programs_available self.refs.make_input_data_available.checked = competition.make_input_data_available $(self.refs.docker_image).val(competition.docker_image)