chore: remove admin jobs (#795)

Impacts: validator

Author: emnoor-reef

validator/app/src/compute_horde_validator/validator/admin.py

@@ -1,6 +1,4 @@
-from compute_horde.base.admin import AddOnlyAdminMixin, ReadOnlyAdminMixin
-from compute_horde.executor_class import EXECUTOR_CLASS
-from django import forms
+from compute_horde.base.admin import ReadOnlyAdminMixin
 from django.contrib import (
     admin,  # noqa
     messages,  # noqa
@@ -11,7 +9,6 @@
 from rangefilter.filters import DateTimeRangeFilter, NumericRangeFilter
 
 from compute_horde_validator.validator.models import (
-    AdminJobRequest,
     Miner,
     MinerBlacklist,
     OrganicJob,
@@ -26,7 +23,6 @@
 from compute_horde_validator.validator.models.scoring.internal import Weights
 
 # noqa
-from compute_horde_validator.validator.tasks import trigger_run_admin_job_request  # noqa
 
 admin.site.site_header = "ComputeHorde Validator Administration"
 admin.site.site_title = "compute_horde_validator"
@@ -35,52 +31,6 @@
 admin.site.index_template = "admin/validator_index.html"
 
 
-class AdminJobRequestForm(forms.ModelForm):
-    executor_class = forms.ChoiceField()
-
-    class Meta:
-        model = AdminJobRequest
-        fields = [
-            "uuid",
-            "miner",
-            "executor_class",
-            "docker_image",
-            "timeout",
-            "args",
-            "use_gpu",
-            "input_url",
-            "output_url",
-            "status_message",
-        ]
-
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        if self.fields:
-            # exclude blacklisted miners from valid results
-            self.fields["miner"].queryset = Miner.objects.exclude(minerblacklist__isnull=False)
-            self.fields["executor_class"].choices = [(name, name) for name in EXECUTOR_CLASS]
-
-
-class AdminJobRequestAddOnlyAdmin(admin.ModelAdmin, AddOnlyAdminMixin):
-    form = AdminJobRequestForm
-    exclude = ["env"]  # not used ?
-    list_display = ["uuid", "executor_class", "docker_image", "use_gpu", "miner", "created_at"]
-    readonly_fields = ["uuid", "status_message"]
-    ordering = ["-created_at"]
-    autocomplete_fields = ["miner"]
-
-    def save_model(self, request, obj, form, change):
-        super().save_model(request, obj, form, change)
-        trigger_run_admin_job_request.delay(obj.id)
-        organic_job = OrganicJob.objects.filter(job_uuid=obj.uuid).first()
-        msg = (
-            f"Please see <a href='/admin/validator/organicjob/{organic_job.pk}/change/'>ORGANIC JOB</a> for further details"
-            if organic_job
-            else f"Job {obj.uuid} failed to initialize"
-        )
-        messages.add_message(request, messages.INFO, mark_safe(msg))
-
-
 class JobReadOnlyAdmin(admin.ModelAdmin, ReadOnlyAdminMixin):
     list_display = ["job_uuid", "miner", "executor_class", "status", "updated_at"]
     search_fields = ["job_uuid", "miner__hotkey"]
@@ -317,7 +267,6 @@ def total_allowance(self, obj):
 admin.site.register(Miner, admin_class=MinerReadOnlyAdmin)
 admin.site.register(OrganicJob, admin_class=JobReadOnlyAdmin)
 admin.site.register(MinerBlacklist, admin_class=MinerBlacklistAdmin)
-admin.site.register(AdminJobRequest, admin_class=AdminJobRequestAddOnlyAdmin)
 admin.site.register(SystemEvent, admin_class=SystemEventAdmin)
 admin.site.register(Weights, admin_class=WeightsReadOnlyAdmin)
 admin.site.register(ValidatorWhitelist, admin_class=ValidatorWhitelistAdmin)

validator/app/src/compute_horde_validator/validator/management/commands/debug_run_organic_job.py

@@ -1,18 +1,29 @@
+import shlex
 import sys
+import uuid
 
 from asgiref.sync import async_to_sync
 from compute_horde.executor_class import DEFAULT_EXECUTOR_CLASS
+from compute_horde.fv_protocol.facilitator_requests import V2JobRequest
 from compute_horde.fv_protocol.validator_requests import JobStatusUpdate
+from compute_horde_core.executor_class import ExecutorClass
+from django.conf import settings
 from django.core.management.base import BaseCommand
-from django.utils import timezone
 
+from compute_horde_validator.validator.allowance.default import allowance
 from compute_horde_validator.validator.models import (
-    AdminJobRequest,
     Miner,
     MinerBlacklist,
     OrganicJob,
 )
-from compute_horde_validator.validator.tasks import run_admin_job_request
+from compute_horde_validator.validator.organic_jobs.miner_client import MinerClient
+from compute_horde_validator.validator.organic_jobs.miner_driver import (
+    drive_organic_job,
+)
+
+
+def get_keypair():
+    return settings.BITTENSOR_WALLET().get_hotkey()
 
 
 async def notify_job_status_update(msg: JobStatusUpdate):
@@ -40,44 +51,36 @@ class Command(BaseCommand):
     """
 
     def add_arguments(self, parser):
-        parser.add_argument("--miner_hotkey", default=None, type=str, help="Miner Hotkey")
-        # TODO: mock miner with address, port, ip_type
-        # parser.add_argument("--miner_address", default=None, type=str, help="Miner IPv4 address")
-        # parser.add_argument("--miner_port", default=None, type=int, help="Miner port")
-        parser.add_argument(
-            "--executor_class", type=str, help="Executor class", default=DEFAULT_EXECUTOR_CLASS
-        )
-        parser.add_argument("--timeout", type=int, help="Timeout value", required=True)
+        parser.add_argument("--miner_hotkey", default=None, help="Miner Hotkey")
+        parser.add_argument("--miner_address", default=None, help="Miner IP address")
+        parser.add_argument("--miner_ip_version", default=4, help="Miner IP version")
+        parser.add_argument("--miner_port", default=None, type=int, help="Miner port")
         parser.add_argument(
-            "--docker_image", type=str, help="docker image for job execution", required=True
+            "--executor_class",
+            type=ExecutorClass,
+            help="Executor class",
+            default=DEFAULT_EXECUTOR_CLASS,
         )
+        parser.add_argument("--docker_image", help="docker image for job execution", required=True)
         parser.add_argument(
-            "--cmd_args",
-            type=str,
-            default="",
-            help="arguments passed to the script or docker image",
+            "--cmd_args", default="", help="arguments passed to the script or docker image"
         )
+        parser.add_argument("--use_gpu", action="store_true", help="use gpu for job execution")
+
         parser.add_argument(
-            "--use_gpu",
-            action="store_true",
-            help="use gpu for job execution",
+            "--download_time_limit", default=10, type=int, help="download time limit in seconds"
         )
         parser.add_argument(
-            "--input_url",
-            type=str,
-            default="",
-            help="input url for job execution",
+            "--execution_time_limit", default=100, type=int, help="execution time limit in seconds"
         )
         parser.add_argument(
-            "--output_url",
-            type=str,
-            default="",
-            help="output url for job execution",
+            "--upload_time_limit", default=10, type=int, help="upload time limit in seconds"
         )
         parser.add_argument(
-            "--nonzero_if_not_complete",
-            action="store_true",
-            help="if job completes with PENDING or FAILED state, exit with non-zero status code",
+            "--streaming_start_time_limit",
+            default=10,
+            type=int,
+            help="streaming start time limit in seconds",
         )
 
     def handle(self, *args, **options):
@@ -102,32 +105,64 @@ def handle(self, *args, **options):
 
         print(f"\nPicked miner: {miner} to run the job")
 
-        job_request = AdminJobRequest.objects.create(
-            miner=miner,
-            timeout=options["timeout"],
+        miner_address = miner.address
+        miner_ip_version = miner.ip_version
+        miner_port = miner.port
+        if options["miner_address"]:
+            miner_address = options["miner_address"]
+            miner_ip_version = options["miner_ip_version"]
+        if options["miner_port"]:
+            miner_port = options["miner_port"]
+
+        job_request = V2JobRequest(
+            uuid=str(uuid.uuid4()),
             executor_class=options["executor_class"],
             docker_image=options["docker_image"],
-            args=options["cmd_args"],
+            args=shlex.split(options["cmd_args"]),
+            env={},
             use_gpu=options["use_gpu"],
-            input_url=options["input_url"],
-            output_url=options["output_url"],
-            created_at=timezone.now(),
+            download_time_limit=options["download_time_limit"],
+            execution_time_limit=options["execution_time_limit"],
+            upload_time_limit=options["upload_time_limit"],
+            streaming_start_time_limit=options["streaming_start_time_limit"],
+        )
+
+        job = OrganicJob.objects.create(
+            job_uuid=str(job_request.uuid),
+            miner=miner,
+            miner_address=miner_address,
+            miner_address_ip_version=miner_ip_version,
+            miner_port=miner_port,
+            namespace=job_request.job_namespace or job_request.docker_image or None,
+            executor_class=job_request.executor_class,
+            job_description="User job from facilitator",
+            block=allowance().get_current_block(),
         )
 
+        async def _run_job():
+            keypair = get_keypair()
+            miner_client = MinerClient(
+                miner_hotkey=miner.hotkey,
+                miner_address=miner_address,
+                miner_port=miner_port,
+                job_uuid=str(job.job_uuid),
+                my_keypair=keypair,
+            )
+            await drive_organic_job(
+                miner_client,
+                job,
+                job_request,
+                notify_callback=notify_job_status_update,
+            )
+
         try:
-            async_to_sync(run_admin_job_request)(job_request.pk, callback=notify_job_status_update)
+            async_to_sync(_run_job)()
+        except Exception as e:
+            print(f"Failed to run job {job.job_uuid}: {e}")
+            sys.exit(1)
         except KeyboardInterrupt:
             print("Interrupted by user")
             sys.exit(1)
 
-        try:
-            job_request.refresh_from_db()
-            job = OrganicJob.objects.get(job_uuid=job_request.uuid)
-            print(f"\nJob {job.job_uuid} done processing")
-        except OrganicJob.DoesNotExist:
-            print(f"\nJob {job_request.uuid} not found")
-            sys.exit(1)
-
-        if options["nonzero_if_not_complete"] and job.status != OrganicJob.Status.COMPLETED:
-            print(f"\nJob {job_request.uuid} was unsuccessful, status = {job.status}")
-            sys.exit(1)
+        job.refresh_from_db()
+        print(f"\nJob {job.job_uuid} done processing with status: {job.status}")

validator/app/src/compute_horde_validator/validator/migrations/0093_delete_adminjobrequest.py

@@ -0,0 +1,15 @@
+# Generated by Django 4.2.19 on 2025-11-05 14:29
+
+from django.db import migrations
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("validator", "0092_remove_prompt_sample_remove_promptsample_series_and_more"),
+    ]
+
+    operations = [
+        migrations.DeleteModel(
+            name="AdminJobRequest",
+        ),
+    ]

validator/app/src/compute_horde_validator/validator/models/spaghetti.py

@@ -1,13 +1,10 @@
 import logging
-import shlex
 import uuid
 from datetime import datetime
 from decimal import Decimal
 from typing import Self
 
 from compute_horde.executor_class import DEFAULT_EXECUTOR_CLASS
-from compute_horde_core.output_upload import OutputUpload, ZipAndHttpPutUpload
-from compute_horde_core.volume import Volume, ZipUrlVolume
 from django.conf import settings
 from django.contrib.postgres.fields import ArrayField
 from django.db import models
@@ -297,48 +294,6 @@ def get_absolute_url(self):
         return f"/admin/validator/organicjob/{self.pk}/change/"
 
 
-class AdminJobRequest(models.Model):
-    uuid = models.UUIDField(default=uuid.uuid4, unique=True)
-    miner = models.ForeignKey(Miner, on_delete=models.PROTECT)
-    executor_class = models.CharField(
-        max_length=255, default=DEFAULT_EXECUTOR_CLASS, help_text="executor hardware class"
-    )
-    timeout = models.PositiveIntegerField(default=300, help_text="timeout in seconds")
-
-    docker_image = models.CharField(max_length=255, help_text="docker image for job execution")
-
-    args = models.TextField(blank=True, help_text="arguments passed to the script or docker image")
-    env = models.JSONField(blank=True, default=dict, help_text="environment variables for the job")
-
-    use_gpu = models.BooleanField(default=False, help_text="Whether to use GPU for the job")
-    input_url = models.URLField(
-        blank=True, help_text="URL to the input data source", max_length=1000
-    )
-    output_url = models.TextField(blank=True, help_text="URL for uploading output")
-
-    created_at = models.DateTimeField(auto_now_add=True)
-
-    status_message = models.TextField(blank=True, default="")
-
-    def get_args(self):
-        return shlex.split(self.args)
-
-    def __str__(self):
-        return f"uuid: {self.uuid} - miner hotkey: {self.miner.hotkey}"
-
-    @property
-    def volume(self) -> Volume | None:
-        if self.input_url:
-            return ZipUrlVolume(contents=self.input_url)
-        return None
-
-    @property
-    def output_upload(self) -> OutputUpload | None:
-        if self.output_url:
-            return ZipAndHttpPutUpload(url=self.output_url)
-        return None
-
-
 class MinerPreliminaryReservationQueryset(models.QuerySet["MinerPreliminaryReservation"]):
     def active(self, at: datetime | None = None) -> Self:
         at = at or now()