Feat: Retry individual eval runs [WIP]

alembic/versions/2026_07_13_add_evaluation_run_attempts.py

@@ -0,0 +1,54 @@
+"""Add evaluation_run_attempts and per-attempt logs
+
+Revision ID: b3f1a9c4d210
+Revises: e7f3a1b2c905
+Create Date: 2026-07-13 00:00:00.000000
+
+"""
+
+from typing import Sequence, Union
+
+from alembic import op
+
+revision: str = "b3f1a9c4d210"
+down_revision: Union[str, Sequence[str], None] = "e7f3a1b2c905"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.execute(
+        """
+        CREATE TABLE evaluation_run_attempts (
+            attempt_id UUID PRIMARY KEY,
+            evaluation_run_id UUID NOT NULL REFERENCES evaluation_runs (evaluation_run_id),
+            attempt_number INT NOT NULL,
+            status evaluationrunstatus NOT NULL,
+            error_code INT,
+            error_message TEXT,
+            cost_usd DOUBLE PRECISION,
+            created_at TIMESTAMPTZ NOT NULL,
+            started_initializing_agent_at TIMESTAMPTZ,
+            started_running_agent_at TIMESTAMPTZ,
+            started_initializing_eval_at TIMESTAMPTZ,
+            started_running_eval_at TIMESTAMPTZ,
+            finished_or_errored_at TIMESTAMPTZ,
+            UNIQUE (evaluation_run_id, attempt_number)
+        )
+        """
+    )
+    op.execute("ALTER TABLE evaluation_run_logs ADD COLUMN attempt_number INT NOT NULL DEFAULT 1;")
+    op.execute("ALTER TABLE evaluation_run_logs DROP CONSTRAINT evaluation_run_logs_pkey;")
+    op.execute(
+        "ALTER TABLE evaluation_run_logs "
+        "ADD CONSTRAINT evaluation_run_logs_pkey PRIMARY KEY (evaluation_run_id, attempt_number, type);"
+    )
+
+
+def downgrade() -> None:
+    op.execute("ALTER TABLE evaluation_run_logs DROP CONSTRAINT evaluation_run_logs_pkey;")
+    op.execute("ALTER TABLE evaluation_run_logs DROP COLUMN attempt_number;")
+    op.execute(
+        "ALTER TABLE evaluation_run_logs ADD CONSTRAINT evaluation_run_logs_pkey PRIMARY KEY (evaluation_run_id, type);"
+    )
+    op.execute("DROP TABLE evaluation_run_attempts;")

api/.env.example

@@ -61,6 +61,7 @@ VALIDATOR_MAX_EVALUATION_RUN_LOG_SIZE_BYTES=500000 # 500 KB
 
 MINER_AGENT_UPLOAD_RATE_LIMIT_SECONDS=86400 # 24 hours
 NUM_EVALS_PER_AGENT=3
+MAX_ATTEMPTS_PER_EVALUATION_RUN=3
 
 SHOULD_RUN_LOOPS=False
 

api/config.py

@@ -187,6 +187,8 @@
     logger.fatal("NUM_EVALS_PER_AGENT is not set in .env")
 NUM_EVALS_PER_AGENT = int(NUM_EVALS_PER_AGENT)
 
+MAX_ATTEMPTS_PER_EVALUATION_RUN = int(os.getenv("MAX_ATTEMPTS_PER_EVALUATION_RUN", "3"))
+
 AGENT_UUID_NAMESPACE = os.getenv("AGENT_UUID_NAMESPACE")
 if not AGENT_UUID_NAMESPACE:
     logger.fatal("AGENT_UUID_NAMESPACE is not set in .env")

api/endpoints/validator.py

@@ -41,7 +41,13 @@
 from db.models import InternalFlagName
 from models.agent import Agent, AgentStatus
 from models.evaluation import Evaluation, EvaluationStatus
-from models.evaluation_run import EvaluationRun, EvaluationRunErrorCode, EvaluationRunLogType, EvaluationRunStatus
+from models.evaluation_run import (
+    EvaluationRun,
+    EvaluationRunErrorCode,
+    EvaluationRunLogType,
+    EvaluationRunStatus,
+    is_retryable_error_code,
+)
 from models.evaluation_set import EvaluationSetGroup
 from models.harbor_task import read_execution_spec_metadata
 from models.openrouter import OpenRouterRuntimeConfig
@@ -75,6 +81,10 @@
     get_evaluation_run_by_id,
     update_evaluation_run_by_id,
 )
+from queries.evaluation_run_attempt import (
+    create_next_attempt_and_reset_evaluation_run,
+    get_attempt_count_for_evaluation_run,
+)
 from queries.internal_flag import get_internal_flags_parsed
 from utils.agent_secrets import AgentKeyDecryptError, AgentKeyEncryptionConfigError, sha256_hex
 from utils.bittensor import validate_signed_timestamp
@@ -682,6 +692,60 @@ async def _maybe_stop_agent_by_score_bound(evaluation: Evaluation) -> bool:
     return stopped
 
 
+async def _maybe_grant_retry(
+    validator: Validator, evaluation_run: EvaluationRun
+) -> Optional[ValidatorUpdateEvaluationRunResponse]:
+    """Grant an in-session retry for an errored run when policy allows it.
+
+    Returns a directive response when a fresh attempt was created, or None to
+    leave the run errored (falling back to the whole-evaluation restart path).
+    Never raises: any failure here just means no retry.
+    """
+
+    if not is_retryable_error_code(int(evaluation_run.error_code) if evaluation_run.error_code is not None else None):
+        return None
+
+    try:
+        attempt_count = await get_attempt_count_for_evaluation_run(evaluation_run.evaluation_run_id)
+        # Legacy runs (created before attempt tracking) have no attempt rows; their
+        # logs already occupy attempt slot 1, so retrying them would collide.
+        if attempt_count == 0:
+            return None
+        if attempt_count >= config.MAX_ATTEMPTS_PER_EVALUATION_RUN:
+            return None
+
+        evaluation = _get_current_evaluation_for_request(validator, evaluation_run.evaluation_id)
+        agent = await get_agent_by_id(evaluation.agent_id)
+        if agent is None or agent.status != _score_bound_active_status(evaluation.evaluation_set_group):
+            return None
+
+        # Mint the fresh upload URL before creating the attempt so a presign
+        # failure aborts cleanly with no dangling pending attempt.
+        artifact_upload_url = await generate_presigned_upload_url(
+            f"artifacts/{evaluation_run.evaluation_run_id}.tar.gz"
+        )
+
+        new_attempt = await create_next_attempt_and_reset_evaluation_run(evaluation_run.evaluation_run_id)
+    except Exception as exc:
+        logger.error(
+            f"Failed to grant retry for evaluation run {evaluation_run.evaluation_run_id}: {type(exc).__name__}: {exc}"
+        )
+        logger.error(traceback.format_exc())
+        return None
+
+    logger.info(f"Granted retry for evaluation run {evaluation_run.evaluation_run_id}")
+    logger.info(f"  Evaluation ID: {evaluation_run.evaluation_id}")
+    logger.info(f"  Problem: {evaluation_run.problem_name}")
+    logger.info(f"  Error code: {evaluation_run.error_code}")
+    logger.info(f"  New attempt number: {new_attempt.attempt_number}")
+
+    return ValidatorUpdateEvaluationRunResponse(
+        retry=True,
+        attempt_number=new_attempt.attempt_number,
+        artifact_upload_url=artifact_upload_url,
+    )
+
+
 async def _is_stale_update_for_score_stopped_evaluation(validator: Validator, evaluation_run: EvaluationRun) -> bool:
     """Return true when a late run update belongs to this validator's already-closed stopped evaluation."""
 
@@ -1030,11 +1094,17 @@ async def validator_update_evaluation_run(
                 )
                 logger.error(traceback.format_exc())
 
+    response = ValidatorUpdateEvaluationRunResponse()
+    if request.updated_status == EvaluationRunStatus.error:
+        retry_directive = await _maybe_grant_retry(validator, evaluation_run)
+        if retry_directive is not None:
+            response = retry_directive
+
     logger.info(f"Validator '{validator.name}' updated an evaluation run")
     logger.info(f"  Evaluation run ID: {request.evaluation_run_id}")
     logger.info(f"  Updated status: {request.updated_status}")
 
-    return ValidatorUpdateEvaluationRunResponse()
+    return response
 
 
 # /validator/disconnect

api/endpoints/validator_models.py

@@ -110,7 +110,11 @@ class ValidatorUpdateEvaluationRunRequest(BaseModel):
 
 
 class ValidatorUpdateEvaluationRunResponse(BaseModel):
-    pass
+    # Set on error updates when the platform wants this run re-executed as a
+    # fresh attempt instead of failing the whole evaluation.
+    retry: bool = False
+    attempt_number: Optional[int] = None
+    artifact_upload_url: Optional[str] = None
 
 
 class ValidatorDisconnectRequest(BaseModel):

db/models/evaluation_run.py

@@ -46,13 +46,40 @@ class EvaluationRun(Base):
     )
 
 
+class EvaluationRunAttempt(Base):
+    __tablename__ = "evaluation_run_attempts"
+
+    attempt_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), primary_key=True)
+    evaluation_run_id: Mapped[UUID] = mapped_column(
+        PG_UUID(as_uuid=True),
+        sa.ForeignKey("evaluation_runs.evaluation_run_id"),
+        nullable=False,
+    )
+    attempt_number: Mapped[int] = mapped_column(sa.Integer, nullable=False)
+    status: Mapped[EvaluationRunStatus] = mapped_column(
+        sa.Enum(EvaluationRunStatus, name="evaluationrunstatus"), nullable=False
+    )
+    error_code: Mapped[Optional[int]] = mapped_column(sa.Integer)
+    error_message: Mapped[Optional[str]] = mapped_column(sa.Text)
+    cost_usd: Mapped[Optional[float]] = mapped_column(sa.Double)
+    created_at: Mapped[datetime] = mapped_column(sa.TIMESTAMP(timezone=True), nullable=False)
+    started_initializing_agent_at: Mapped[Optional[datetime]] = mapped_column(sa.TIMESTAMP(timezone=True))
+    started_running_agent_at: Mapped[Optional[datetime]] = mapped_column(sa.TIMESTAMP(timezone=True))
+    started_initializing_eval_at: Mapped[Optional[datetime]] = mapped_column(sa.TIMESTAMP(timezone=True))
+    started_running_eval_at: Mapped[Optional[datetime]] = mapped_column(sa.TIMESTAMP(timezone=True))
+    finished_or_errored_at: Mapped[Optional[datetime]] = mapped_column(sa.TIMESTAMP(timezone=True))
+
+    __table_args__ = (sa.UniqueConstraint("evaluation_run_id", "attempt_number"),)
+
+
 class EvaluationRunLog(Base):
     __tablename__ = "evaluation_run_logs"
 
     evaluation_run_id: Mapped[UUID] = mapped_column(PG_UUID(as_uuid=True), nullable=False)
+    attempt_number: Mapped[int] = mapped_column(sa.Integer, nullable=False, server_default="1")
     logs: Mapped[Optional[str]] = mapped_column(sa.Text)
     type: Mapped[Optional[EvaluationRunLogType]] = mapped_column(
         sa.Enum(EvaluationRunLogType, name="evaluationrunlogtype")
     )
 
-    __table_args__ = (sa.PrimaryKeyConstraint("evaluation_run_id", "type"),)
+    __table_args__ = (sa.PrimaryKeyConstraint("evaluation_run_id", "attempt_number", "type"),)

execution/engine.py

@@ -26,12 +26,15 @@
 _JOB_NAME_FORMAT = "{problem_name}__{evaluation_run_id}"
 
 
-def _format_job_name(problem_name: str, evaluation_run_id: UUID) -> str:
-    """Format the job name Harbor uses for one evaluation run."""
-    return _JOB_NAME_FORMAT.format(
+def _format_job_name(problem_name: str, evaluation_run_id: UUID, attempt_number: int = 1) -> str:
+    """Format the job name Harbor uses for one evaluation run attempt."""
+    job_name = _JOB_NAME_FORMAT.format(
         problem_name=problem_name,
         evaluation_run_id=evaluation_run_id,
     )
+    if attempt_number > 1:
+        job_name = f"{job_name}__a{attempt_number}"
+    return job_name
 
 
 @contextmanager
@@ -100,6 +103,7 @@ async def evaluate(
         fetch_task_url: Callable[[str], Awaitable[str]] | None = None,
         on_agent_started: Callable[[], Awaitable[None]] | None = None,
         on_verification_started: Callable[[TrialSnapshot], Awaitable[None]] | None = None,
+        attempt_number: int = 1,
     ) -> ExecutionResult:
         """Run the task referenced by the evaluation set and normalize the result.
 
@@ -117,6 +121,7 @@ async def evaluate(
             parsed_spec=parsed_spec,
             task_dir=task_dir,
             problem_name=problem_name,
+            attempt_number=attempt_number,
         )
         inference_seed = problem_seed(problem_name)
 
@@ -204,9 +209,12 @@ def _build_run_request(
         parsed_spec: HarborRemoteTaskExecutionSpec,
         task_dir: Path,
         problem_name: str,
+        attempt_number: int = 1,
     ) -> ExecutionRunRequest:
         """Bundle the resolved inputs needed to invoke the Harbor runner."""
-        job_name = _format_job_name(problem_name=problem_name, evaluation_run_id=evaluation_run_id)
+        job_name = _format_job_name(
+            problem_name=problem_name, evaluation_run_id=evaluation_run_id, attempt_number=attempt_number
+        )
         results_dir = Path(self.results_dir or DEFAULT_RESULTS_DIR).expanduser().resolve()
 
         # If the Engine specifies a max agent timeout, compare it with the task

inference_gateway/cost_hash_map.py

@@ -6,7 +6,7 @@
 #            information is tracked here in memory.
 
 import time
-from uuid import UUID
+from typing import Hashable
 
 from pydantic import BaseModel
 
@@ -33,22 +33,22 @@ def _cleanup(self):
             }
             self.last_cleanup_at = now
 
-    def get_cost(self, uuid: UUID) -> float:
+    def get_cost(self, key: Hashable) -> float:
         self._cleanup()
 
-        if uuid in self.cost_hash_map:
-            self.cost_hash_map[uuid].last_accessed_at = time.time()
-            return self.cost_hash_map[uuid].cost
+        if key in self.cost_hash_map:
+            self.cost_hash_map[key].last_accessed_at = time.time()
+            return self.cost_hash_map[key].cost
         else:
             # TODO ADAM: db
             return 0
 
-    def add_cost(self, uuid: UUID, cost: float):
+    def add_cost(self, key: Hashable, cost: float):
         self._cleanup()
 
-        if uuid in self.cost_hash_map:
-            entry = self.cost_hash_map[uuid]
+        if key in self.cost_hash_map:
+            entry = self.cost_hash_map[key]
             entry.cost += cost
             entry.last_accessed_at = time.time()
         else:
-            self.cost_hash_map[uuid] = CostHashMapEntry(cost=cost, last_accessed_at=time.time())
+            self.cost_hash_map[key] = CostHashMapEntry(cost=cost, last_accessed_at=time.time())