diff --git a/alembic/versions/2026_07_13_add_evaluation_run_attempts.py b/alembic/versions/2026_07_13_add_evaluation_run_attempts.py new file mode 100644 index 00000000..259a1c58 --- /dev/null +++ b/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;") diff --git a/api/.env.example b/api/.env.example index 1335e5dc..21c65bde 100644 --- a/api/.env.example +++ b/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 diff --git a/api/config.py b/api/config.py index 46f51840..12bd559f 100644 --- a/api/config.py +++ b/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") diff --git a/api/endpoints/validator.py b/api/endpoints/validator.py index 9a3476ed..c657b9c3 100644 --- a/api/endpoints/validator.py +++ b/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 diff --git a/api/endpoints/validator_models.py b/api/endpoints/validator_models.py index 122d7aa6..0229c321 100644 --- a/api/endpoints/validator_models.py +++ b/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): diff --git a/db/models/evaluation_run.py b/db/models/evaluation_run.py index b600b03c..31a25b89 100644 --- a/db/models/evaluation_run.py +++ b/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"),) diff --git a/execution/engine.py b/execution/engine.py index 7a28f805..e3b71d7c 100644 --- a/execution/engine.py +++ b/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 diff --git a/inference_gateway/cost_hash_map.py b/inference_gateway/cost_hash_map.py index a7ce61af..c7f094cf 100644 --- a/inference_gateway/cost_hash_map.py +++ b/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()) diff --git a/inference_gateway/main.py b/inference_gateway/main.py index 91f1c9f0..07740ded 100644 --- a/inference_gateway/main.py +++ b/inference_gateway/main.py @@ -26,7 +26,7 @@ from inference_gateway.providers.targon import TargonProvider from models.evaluation_run import EvaluationRunStatus from queries.embedding import create_new_embedding, update_embedding_by_id -from queries.evaluation_run import get_evaluation_run_status_by_id +from queries.evaluation_run import get_evaluation_run_status_and_attempt_by_id from queries.inference import create_new_inference, update_inference_by_id from utils.database import deinitialize_database, get_debug_query_info, initialize_database from utils.logger import setup_logging @@ -127,17 +127,20 @@ async def inference(request: InferenceRequest) -> InferenceResponse: status_code=422, detail="If you specify a tool mode of REQUIRED, you must specify at least one tool." ) + cost_key = (request.evaluation_run_id, 1) if config.USE_DATABASE and config.CHECK_EVALUATION_RUNS: - # Get the status of the evaluation run - evaluation_run_status = await get_evaluation_run_status_by_id(request.evaluation_run_id) + status_and_attempt = await get_evaluation_run_status_and_attempt_by_id(request.evaluation_run_id) # Make sure the evaluation run actually exists - if evaluation_run_status is None: + if status_and_attempt is None: raise HTTPException( status_code=400, detail=f"No evaluation run exists with the given evaluation run ID {request.evaluation_run_id}.", ) + evaluation_run_status, attempt_number = status_and_attempt + cost_key = (request.evaluation_run_id, attempt_number) + # Make sure the evaluation run is in the running_agent state if evaluation_run_status != EvaluationRunStatus.running_agent: raise HTTPException( @@ -146,7 +149,7 @@ async def inference(request: InferenceRequest) -> InferenceResponse: ) # Make sure the evaluation run has not reached or exceeded its cost limit - cost = cost_hash_map.get_cost(request.evaluation_run_id) + cost = cost_hash_map.get_cost(cost_key) if cost >= config.MAX_COST_PER_EVALUATION_RUN_USD: raise HTTPException( status_code=429, @@ -188,7 +191,7 @@ async def inference(request: InferenceRequest) -> InferenceResponse: ) if response.cost_usd is not None: - cost_hash_map.add_cost(request.evaluation_run_id, response.cost_usd) + cost_hash_map.add_cost(cost_key, response.cost_usd) if response.status_code == 200: return InferenceResponse(content=response.content, tool_calls=response.tool_calls) @@ -202,17 +205,20 @@ async def inference(request: InferenceRequest) -> InferenceResponse: @app.post("/api/embedding") @handle_http_exceptions async def embedding(request: EmbeddingRequest) -> EmbeddingResponse: + cost_key = (request.evaluation_run_id, 1) if config.USE_DATABASE and config.CHECK_EVALUATION_RUNS: - # Get the status of the evaluation run - evaluation_run_status = await get_evaluation_run_status_by_id(request.evaluation_run_id) + status_and_attempt = await get_evaluation_run_status_and_attempt_by_id(request.evaluation_run_id) # Make sure the evaluation run actually exists - if evaluation_run_status is None: + if status_and_attempt is None: raise HTTPException( status_code=400, detail=f"No evaluation run exists with the given evaluation run ID {request.evaluation_run_id}.", ) + evaluation_run_status, attempt_number = status_and_attempt + cost_key = (request.evaluation_run_id, attempt_number) + # Make sure the evaluation run is in the running_agent state if evaluation_run_status != EvaluationRunStatus.running_agent: raise HTTPException( @@ -221,7 +227,7 @@ async def embedding(request: EmbeddingRequest) -> EmbeddingResponse: ) # Make sure the evaluation run has not reached or exceeded its cost limit - cost = cost_hash_map.get_cost(request.evaluation_run_id) + cost = cost_hash_map.get_cost(cost_key) if cost >= config.MAX_COST_PER_EVALUATION_RUN_USD: raise HTTPException( status_code=429, @@ -255,7 +261,7 @@ async def embedding(request: EmbeddingRequest) -> EmbeddingResponse: ) if response.cost_usd is not None: - cost_hash_map.add_cost(request.evaluation_run_id, response.cost_usd) + cost_hash_map.add_cost(cost_key, response.cost_usd) if response.status_code == 200: return EmbeddingResponse(embedding=response.embedding) @@ -272,7 +278,13 @@ class UsageResponse(BaseModel): @app.get("/api/usage") @handle_http_exceptions async def usage(evaluation_run_id: UUID) -> UsageResponse: - used_cost_usd = cost_hash_map.get_cost(evaluation_run_id) + attempt_number = 1 + if config.USE_DATABASE and config.CHECK_EVALUATION_RUNS: + status_and_attempt = await get_evaluation_run_status_and_attempt_by_id(evaluation_run_id) + if status_and_attempt is not None: + attempt_number = status_and_attempt[1] + + used_cost_usd = cost_hash_map.get_cost((evaluation_run_id, attempt_number)) return UsageResponse( used_cost_usd=used_cost_usd, remaining_cost_usd=config.MAX_COST_PER_EVALUATION_RUN_USD - used_cost_usd, diff --git a/models/evaluation_run.py b/models/evaluation_run.py index a8608cf6..8dc3e293 100644 --- a/models/evaluation_run.py +++ b/models/evaluation_run.py @@ -94,6 +94,35 @@ def is_platform_error(self) -> bool: return 3000 <= self.value < 4000 +# Non-agent errors that must never trigger an in-session retry: score-bound +# pruning is terminal by definition, PLATFORM_RESTARTED_* are only set by bulk +# cleanup after the owning session is already gone, and an unknown problem +# fails identically on every attempt. +NON_RETRYABLE_PLATFORM_ERROR_CODES = frozenset( + { + EvaluationRunErrorCode.VALIDATOR_UNKNOWN_PROBLEM.value, + EvaluationRunErrorCode.PLATFORM_RESTARTED_WHILE_PENDING.value, + EvaluationRunErrorCode.PLATFORM_RESTARTED_WHILE_INIT_AGENT.value, + EvaluationRunErrorCode.PLATFORM_RESTARTED_WHILE_RUNNING_AGENT.value, + EvaluationRunErrorCode.PLATFORM_RESTARTED_WHILE_INIT_EVAL.value, + EvaluationRunErrorCode.PLATFORM_RESTARTED_WHILE_RUNNING_EVAL.value, + EvaluationRunErrorCode.PLATFORM_PRUNED_BY_SCORE_BOUND.value, + } +) + + +def is_retryable_error_code(error_code: int | None) -> bool: + """Whether an errored run may be retried in-session with a fresh attempt. + + Agent errors (1xxx) already count toward evaluation success and are never + retried; only validator/platform faults outside the denylist qualify. + """ + if error_code is None: + return False + code = int(error_code) + return 2000 <= code < 4000 and code not in NON_RETRYABLE_PLATFORM_ERROR_CODES + + class EvaluationRunStatus(str, Enum): """Lifecycle states for one problem inside an evaluation.""" @@ -106,6 +135,26 @@ class EvaluationRunStatus(str, Enum): error = "error" +class EvaluationRunAttempt(BaseModel): + """One attempt at executing an evaluation run. The evaluation_runs row mirrors the latest attempt.""" + + attempt_id: UUID + evaluation_run_id: UUID + attempt_number: int + status: EvaluationRunStatus + + error_code: Optional[int] = None + error_message: Optional[str] = None + cost_usd: Optional[float] = None + + created_at: datetime + started_initializing_agent_at: Optional[datetime] = None + started_running_agent_at: Optional[datetime] = None + started_initializing_eval_at: Optional[datetime] = None + started_running_eval_at: Optional[datetime] = None + finished_or_errored_at: Optional[datetime] = None + + class EvaluationRun(BaseModel): """Persisted state for one problem run inside a broader evaluation.""" @@ -141,6 +190,7 @@ class EvaluationRunDetail(EvaluationRun): problem_total_runs: int | None = None problem_average_time_seconds: float | None = None problem_average_cost_usd: float | None = None + attempt_count: int = 1 class EvaluationRunLogType(str, Enum): diff --git a/queries/evaluation.py b/queries/evaluation.py index 5e7edc34..b7e8fe9c 100644 --- a/queries/evaluation.py +++ b/queries/evaluation.py @@ -15,6 +15,25 @@ logger = logging.getLogger(__name__) +# Copies terminal error state from just-updated evaluation_runs mirrors onto their +# current attempts. $1 narrowing differs per caller; see usages below. +_MIRROR_ERRORED_RUNS_TO_CURRENT_ATTEMPTS = """ + UPDATE evaluation_run_attempts era + SET + status = er.status, + error_code = er.error_code, + error_message = er.error_message, + finished_or_errored_at = er.finished_or_errored_at + FROM evaluation_runs er + WHERE er.evaluation_run_id = era.evaluation_run_id + AND er.status = 'error' + AND era.attempt_number = ( + SELECT MAX(a2.attempt_number) FROM evaluation_run_attempts a2 + WHERE a2.evaluation_run_id = era.evaluation_run_id + ) + AND era.status NOT IN ('finished', 'error') +""" + @dataclass(slots=True, frozen=True) class LocalEvaluationScoreBound: @@ -410,74 +429,91 @@ async def transition_agent_status_if_matches( async def set_unfinished_evaluation_runs_to_score_pruned( conn: DatabaseConnection, evaluation_id: UUID, error_message: str ) -> int: - return await conn.fetchval( - f""" - WITH updated AS ( + async with conn.conn.transaction(): + count = await conn.fetchval( + f""" + WITH updated AS ( + UPDATE evaluation_runs + SET + status = '{EvaluationRunStatus.error.value}', + error_code = {EvaluationRunErrorCode.PLATFORM_PRUNED_BY_SCORE_BOUND.value}, + error_message = $2, + finished_or_errored_at = NOW() + WHERE evaluation_id = $1 + AND status NOT IN ( + '{EvaluationRunStatus.finished.value}'::evaluationrunstatus, + '{EvaluationRunStatus.error.value}'::evaluationrunstatus + ) + RETURNING 1 + ) + SELECT COUNT(*)::int FROM updated + """, + evaluation_id, + error_message, + ) + + await conn.execute( + _MIRROR_ERRORED_RUNS_TO_CURRENT_ATTEMPTS + " AND er.evaluation_id = $1", + evaluation_id, + ) + + return count + + +@db_operation +async def set_all_unfinished_evaluation_runs_to_errored(conn: DatabaseConnection, error_message: str) -> None: + async with conn.conn.transaction(): + await conn.execute( + f""" UPDATE evaluation_runs SET status = '{EvaluationRunStatus.error.value}', - error_code = {EvaluationRunErrorCode.PLATFORM_PRUNED_BY_SCORE_BOUND.value}, - error_message = $2, + error_code = CASE + WHEN status = '{EvaluationRunStatus.pending.value}' THEN {EvaluationRunErrorCode.VALIDATOR_FAILED_PENDING.value} + WHEN status = '{EvaluationRunStatus.initializing_agent.value}' THEN {EvaluationRunErrorCode.VALIDATOR_FAILED_INIT_AGENT.value} + WHEN status = '{EvaluationRunStatus.running_agent.value}' THEN {EvaluationRunErrorCode.VALIDATOR_FAILED_RUNNING_AGENT.value} + WHEN status = '{EvaluationRunStatus.initializing_eval.value}' THEN {EvaluationRunErrorCode.VALIDATOR_FAILED_INIT_EVAL.value} + WHEN status = '{EvaluationRunStatus.running_eval.value}' THEN {EvaluationRunErrorCode.VALIDATOR_FAILED_RUNNING_EVAL.value} + ELSE {EvaluationRunErrorCode.VALIDATOR_INTERNAL_ERROR.value} + END, + error_message = $1, finished_or_errored_at = NOW() - WHERE evaluation_id = $1 - AND status NOT IN ( - '{EvaluationRunStatus.finished.value}'::evaluationrunstatus, - '{EvaluationRunStatus.error.value}'::evaluationrunstatus - ) - RETURNING 1 + WHERE + status NOT IN ('{EvaluationRunStatus.finished.value}', '{EvaluationRunStatus.error.value}') + """, + error_message, ) - SELECT COUNT(*)::int FROM updated - """, - evaluation_id, - error_message, - ) - -@db_operation -async def set_all_unfinished_evaluation_runs_to_errored(conn: DatabaseConnection, error_message: str) -> None: - await conn.execute( - f""" - UPDATE evaluation_runs - SET - status = '{EvaluationRunStatus.error.value}', - error_code = CASE - WHEN status = '{EvaluationRunStatus.pending.value}' THEN {EvaluationRunErrorCode.VALIDATOR_FAILED_PENDING.value} - WHEN status = '{EvaluationRunStatus.initializing_agent.value}' THEN {EvaluationRunErrorCode.VALIDATOR_FAILED_INIT_AGENT.value} - WHEN status = '{EvaluationRunStatus.running_agent.value}' THEN {EvaluationRunErrorCode.VALIDATOR_FAILED_RUNNING_AGENT.value} - WHEN status = '{EvaluationRunStatus.initializing_eval.value}' THEN {EvaluationRunErrorCode.VALIDATOR_FAILED_INIT_EVAL.value} - WHEN status = '{EvaluationRunStatus.running_eval.value}' THEN {EvaluationRunErrorCode.VALIDATOR_FAILED_RUNNING_EVAL.value} - ELSE {EvaluationRunErrorCode.VALIDATOR_INTERNAL_ERROR.value} - END, - error_message = $1, - finished_or_errored_at = NOW() - WHERE - status NOT IN ('{EvaluationRunStatus.finished.value}', '{EvaluationRunStatus.error.value}') - """, - error_message, - ) + await conn.execute(_MIRROR_ERRORED_RUNS_TO_CURRENT_ATTEMPTS) @db_operation async def update_unfinished_evaluation_runs_in_evaluation_id_to_errored( conn: DatabaseConnection, evaluation_id: UUID, error_message: str ) -> None: - await conn.execute( - f""" - UPDATE evaluation_runs - SET - status = '{EvaluationRunStatus.error.value}', - error_code = CASE - WHEN status = '{EvaluationRunStatus.pending.value}' THEN {EvaluationRunErrorCode.PLATFORM_RESTARTED_WHILE_PENDING.value} - WHEN status = '{EvaluationRunStatus.initializing_agent.value}' THEN {EvaluationRunErrorCode.PLATFORM_RESTARTED_WHILE_INIT_AGENT.value} - WHEN status = '{EvaluationRunStatus.running_agent.value}' THEN {EvaluationRunErrorCode.PLATFORM_RESTARTED_WHILE_RUNNING_AGENT.value} - WHEN status = '{EvaluationRunStatus.initializing_eval.value}' THEN {EvaluationRunErrorCode.PLATFORM_RESTARTED_WHILE_INIT_EVAL.value} - WHEN status = '{EvaluationRunStatus.running_eval.value}' THEN {EvaluationRunErrorCode.PLATFORM_RESTARTED_WHILE_RUNNING_EVAL.value} - END, - error_message = $2, - finished_or_errored_at = NOW() - WHERE evaluation_id = $1 - AND status NOT IN ('{EvaluationRunStatus.finished.value}', '{EvaluationRunStatus.error.value}') - """, - evaluation_id, - error_message, - ) + async with conn.conn.transaction(): + await conn.execute( + f""" + UPDATE evaluation_runs + SET + status = '{EvaluationRunStatus.error.value}', + error_code = CASE + WHEN status = '{EvaluationRunStatus.pending.value}' THEN {EvaluationRunErrorCode.PLATFORM_RESTARTED_WHILE_PENDING.value} + WHEN status = '{EvaluationRunStatus.initializing_agent.value}' THEN {EvaluationRunErrorCode.PLATFORM_RESTARTED_WHILE_INIT_AGENT.value} + WHEN status = '{EvaluationRunStatus.running_agent.value}' THEN {EvaluationRunErrorCode.PLATFORM_RESTARTED_WHILE_RUNNING_AGENT.value} + WHEN status = '{EvaluationRunStatus.initializing_eval.value}' THEN {EvaluationRunErrorCode.PLATFORM_RESTARTED_WHILE_INIT_EVAL.value} + WHEN status = '{EvaluationRunStatus.running_eval.value}' THEN {EvaluationRunErrorCode.PLATFORM_RESTARTED_WHILE_RUNNING_EVAL.value} + END, + error_message = $2, + finished_or_errored_at = NOW() + WHERE evaluation_id = $1 + AND status NOT IN ('{EvaluationRunStatus.finished.value}', '{EvaluationRunStatus.error.value}') + """, + evaluation_id, + error_message, + ) + + await conn.execute( + _MIRROR_ERRORED_RUNS_TO_CURRENT_ATTEMPTS + " AND er.evaluation_id = $1", + evaluation_id, + ) diff --git a/queries/evaluation_run.py b/queries/evaluation_run.py index f599129a..d755faa1 100644 --- a/queries/evaluation_run.py +++ b/queries/evaluation_run.py @@ -35,20 +35,29 @@ async def get_evaluation_run_by_id(conn: DatabaseConnection, evaluation_run_id: @db_operation -async def get_evaluation_run_status_by_id( +async def get_evaluation_run_status_and_attempt_by_id( conn: DatabaseConnection, evaluation_run_id: UUID -) -> Optional[EvaluationRunStatus]: - status = await conn.fetchval( +) -> Optional[tuple[EvaluationRunStatus, int]]: + """Return the run's mirrored status and its current attempt number (1 for legacy runs).""" + row = await conn.fetchrow( """ - SELECT status FROM evaluation_runs WHERE evaluation_run_id = $1 + SELECT + er.status, + COALESCE( + (SELECT MAX(era.attempt_number) FROM evaluation_run_attempts era + WHERE era.evaluation_run_id = er.evaluation_run_id), + 1 + )::int AS attempt_number + FROM evaluation_runs er + WHERE er.evaluation_run_id = $1 """, evaluation_run_id, ) - if status is None: + if row is None: return None - return EvaluationRunStatus(status) + return EvaluationRunStatus(row["status"]), row["attempt_number"] @db_operation @@ -73,6 +82,7 @@ def _parse_metrics_row(row: asyncpg.Record) -> dict: "problem_total_runs": row["problem_total_runs"], "problem_average_time_seconds": row["problem_average_time_seconds"], "problem_average_cost_usd": row["problem_average_cost_usd"], + "attempt_count": row["attempt_count"], } @@ -94,7 +104,12 @@ async def _get_evaluation_run_metrics_by_ids( AND er.finished_or_errored_at IS NOT NULL THEN EXTRACT(EPOCH FROM (er.finished_or_errored_at - er.started_initializing_agent_at)) ELSE NULL - END AS run_time_seconds + END AS run_time_seconds, + GREATEST( + (SELECT COUNT(*) FROM evaluation_run_attempts era + WHERE era.evaluation_run_id = er.evaluation_run_id), + 1 + )::int AS attempt_count FROM evaluation_runs er JOIN evaluations e ON e.evaluation_id = er.evaluation_id WHERE er.evaluation_run_id = ANY($1::uuid[]) @@ -150,6 +165,7 @@ async def _get_evaluation_run_metrics_by_ids( SELECT tr.evaluation_run_id, tr.run_time_seconds, + tr.attempt_count, COALESCE(pa.problem_total_runs, 0) AS problem_total_runs, pa.problem_average_time_seconds, pa.problem_average_cost_usd @@ -179,65 +195,107 @@ async def get_evaluation_run_metrics_by_id(conn: DatabaseConnection, evaluation_ @db_operation async def update_evaluation_run_by_id(conn: DatabaseConnection, evaluation_run: EvaluationRun) -> None: - await conn.execute( - """ - UPDATE evaluation_runs SET - status = $2, - patch = $3, - test_results = $4, - verifier_reward = $5, - error_code = $6, - error_message = $7, - started_initializing_agent_at = $8, - started_running_agent_at = $9, - started_initializing_eval_at = $10, - started_running_eval_at = $11, - finished_or_errored_at = $12, - cost_usd = $13 - WHERE evaluation_run_id = $1 - """, - evaluation_run.evaluation_run_id, - evaluation_run.status.value, - evaluation_run.patch, - json.dumps( - [ - test_result.model_dump(exclude={"test_alias"}, exclude_none=True) - for test_result in evaluation_run.test_results - ] + async with conn.conn.transaction(): + await conn.execute( + """ + UPDATE evaluation_runs SET + status = $2, + patch = $3, + test_results = $4, + verifier_reward = $5, + error_code = $6, + error_message = $7, + started_initializing_agent_at = $8, + started_running_agent_at = $9, + started_initializing_eval_at = $10, + started_running_eval_at = $11, + finished_or_errored_at = $12, + cost_usd = $13 + WHERE evaluation_run_id = $1 + """, + evaluation_run.evaluation_run_id, + evaluation_run.status.value, + evaluation_run.patch, + json.dumps( + [ + test_result.model_dump(exclude={"test_alias"}, exclude_none=True) + for test_result in evaluation_run.test_results + ] + ) + if evaluation_run.test_results is not None + else None, + evaluation_run.verifier_reward, + evaluation_run.error_code, + evaluation_run.error_message, + evaluation_run.started_initializing_agent_at, + evaluation_run.started_running_agent_at, + evaluation_run.started_initializing_eval_at, + evaluation_run.started_running_eval_at, + evaluation_run.finished_or_errored_at, + evaluation_run.cost_usd, + ) + + # Mirror lifecycle fields onto the current (highest-numbered) attempt. + # No-op for legacy runs that have no attempt rows. + await conn.execute( + """ + UPDATE evaluation_run_attempts SET + status = $2, + error_code = $3, + error_message = $4, + cost_usd = $5, + started_initializing_agent_at = $6, + started_running_agent_at = $7, + started_initializing_eval_at = $8, + started_running_eval_at = $9, + finished_or_errored_at = $10 + WHERE evaluation_run_id = $1 + AND attempt_number = ( + SELECT MAX(attempt_number) FROM evaluation_run_attempts WHERE evaluation_run_id = $1 + ) + """, + evaluation_run.evaluation_run_id, + evaluation_run.status.value, + int(evaluation_run.error_code) if evaluation_run.error_code is not None else None, + evaluation_run.error_message, + evaluation_run.cost_usd, + evaluation_run.started_initializing_agent_at, + evaluation_run.started_running_agent_at, + evaluation_run.started_initializing_eval_at, + evaluation_run.started_running_eval_at, + evaluation_run.finished_or_errored_at, ) - if evaluation_run.test_results is not None - else None, - evaluation_run.verifier_reward, - evaluation_run.error_code, - evaluation_run.error_message, - evaluation_run.started_initializing_agent_at, - evaluation_run.started_running_agent_at, - evaluation_run.started_initializing_eval_at, - evaluation_run.started_running_eval_at, - evaluation_run.finished_or_errored_at, - evaluation_run.cost_usd, - ) @db_operation async def create_evaluation_run(conn: DatabaseConnection, evaluation_id: UUID, problem_name: str) -> UUID: evaluation_run_id = uuid4() - await conn.execute( - """ - INSERT INTO evaluation_runs ( + async with conn.conn.transaction(): + await conn.execute( + """ + INSERT INTO evaluation_runs ( + evaluation_run_id, + evaluation_id, + problem_name, + status, + created_at + ) VALUES ($1, $2, $3, $4, NOW()) + """, evaluation_run_id, evaluation_id, problem_name, - status, - created_at - ) VALUES ($1, $2, $3, $4, NOW()) - """, - evaluation_run_id, - evaluation_id, - problem_name, - EvaluationRunStatus.pending.value, - ) + EvaluationRunStatus.pending.value, + ) + await conn.execute( + """ + INSERT INTO evaluation_run_attempts (attempt_id, evaluation_run_id, attempt_number, status, created_at) + VALUES ($1, $2, 1, $3, NOW()) + """, + uuid4(), + evaluation_run_id, + EvaluationRunStatus.pending.value, + ) logger.debug( f"Created evaluation run {evaluation_run_id} for evaluation {evaluation_id} with problem name {problem_name}" @@ -250,30 +308,40 @@ async def create_evaluation_run(conn: DatabaseConnection, evaluation_id: UUID, p async def create_evaluation_runs( conn: DatabaseConnection, evaluation_id: UUID, evaluation_set_problems: List[EvaluationSetProblem] ) -> None: - await conn.executemany( - """ - INSERT INTO evaluation_runs ( - evaluation_run_id, + run_rows = [ + ( + uuid4(), evaluation_id, - problem_name, - benchmark_family, - execution_spec, - status, - created_at - ) VALUES ($1, $2, $3, $4, $5::jsonb, $6, NOW()) - """, - [ - ( - uuid4(), + problem.problem_name, + problem.benchmark_family, + json.dumps(problem.execution_spec) if problem.execution_spec is not None else None, + EvaluationRunStatus.pending.value, + ) + for problem in evaluation_set_problems + ] + + async with conn.conn.transaction(): + await conn.executemany( + """ + INSERT INTO evaluation_runs ( + evaluation_run_id, evaluation_id, - problem.problem_name, - problem.benchmark_family, - json.dumps(problem.execution_spec) if problem.execution_spec is not None else None, - EvaluationRunStatus.pending.value, - ) - for problem in evaluation_set_problems - ], - ) + problem_name, + benchmark_family, + execution_spec, + status, + created_at + ) VALUES ($1, $2, $3, $4, $5::jsonb, $6, NOW()) + """, + run_rows, + ) + await conn.executemany( + """ + INSERT INTO evaluation_run_attempts (attempt_id, evaluation_run_id, attempt_number, status, created_at) + VALUES ($1, $2, 1, $3, NOW()) + """, + [(uuid4(), row[0], EvaluationRunStatus.pending.value) for row in run_rows], + ) logger.debug(f"Created {len(evaluation_set_problems)} evaluation runs for evaluation {evaluation_id}") @@ -286,9 +354,17 @@ async def create_evaluation_run_log( """ INSERT INTO evaluation_run_logs ( evaluation_run_id, + attempt_number, type, logs - ) VALUES ($1, $2, $3) + ) VALUES ( + $1, + COALESCE( + (SELECT MAX(attempt_number) FROM evaluation_run_attempts WHERE evaluation_run_id = $1), 1 + ), + $2, + $3 + ) """, evaluation_run_id, type, @@ -309,7 +385,11 @@ async def check_if_evaluation_run_logs_exist( """ SELECT EXISTS ( SELECT 1 FROM evaluation_run_logs - WHERE evaluation_run_id = $1 AND type = $2 + WHERE evaluation_run_id = $1 + AND type = $2 + AND attempt_number = COALESCE( + (SELECT MAX(attempt_number) FROM evaluation_run_attempts WHERE evaluation_run_id = $1), 1 + ) ) """, evaluation_run_id, @@ -326,6 +406,8 @@ async def get_evaluation_run_logs_by_id( SELECT logs FROM evaluation_run_logs WHERE type = $1 and evaluation_run_id = $2 + ORDER BY attempt_number DESC + LIMIT 1 """, type, evaluation_run_id, diff --git a/queries/evaluation_run_attempt.py b/queries/evaluation_run_attempt.py new file mode 100644 index 00000000..9486b3f2 --- /dev/null +++ b/queries/evaluation_run_attempt.py @@ -0,0 +1,88 @@ +import logging +from typing import List +from uuid import UUID, uuid4 + +import asyncpg + +from models.evaluation_run import EvaluationRunAttempt, EvaluationRunStatus +from utils.database import DatabaseConnection, db_operation + +logger = logging.getLogger(__name__) + + +def _parse_attempt_from_row(row: asyncpg.Record) -> EvaluationRunAttempt: + return EvaluationRunAttempt(**dict(row)) + + +@db_operation +async def get_attempts_for_evaluation_run( + conn: DatabaseConnection, evaluation_run_id: UUID +) -> List[EvaluationRunAttempt]: + rows = await conn.fetch( + """ + SELECT * + FROM evaluation_run_attempts + WHERE evaluation_run_id = $1 + ORDER BY attempt_number ASC + """, + evaluation_run_id, + ) + + return [_parse_attempt_from_row(row) for row in rows] + + +@db_operation +async def get_attempt_count_for_evaluation_run(conn: DatabaseConnection, evaluation_run_id: UUID) -> int: + return await conn.fetchval( + """ + SELECT COUNT(*)::int + FROM evaluation_run_attempts + WHERE evaluation_run_id = $1 + """, + evaluation_run_id, + ) + + +@db_operation +async def create_next_attempt_and_reset_evaluation_run( + conn: DatabaseConnection, evaluation_run_id: UUID +) -> EvaluationRunAttempt: + """Insert attempt N+1 as pending and reset the evaluation_runs mirror, atomically.""" + + async with conn.conn.transaction(): + row = await conn.fetchrow( + """ + INSERT INTO evaluation_run_attempts (attempt_id, evaluation_run_id, attempt_number, status, created_at) + SELECT $2, $1, COALESCE(MAX(attempt_number), 0) + 1, $3, NOW() + FROM evaluation_run_attempts + WHERE evaluation_run_id = $1 + RETURNING * + """, + evaluation_run_id, + uuid4(), + EvaluationRunStatus.pending.value, + ) + await conn.execute( + """ + UPDATE evaluation_runs SET + status = $2, + patch = NULL, + test_results = NULL, + verifier_reward = NULL, + error_code = NULL, + error_message = NULL, + cost_usd = NULL, + started_initializing_agent_at = NULL, + started_running_agent_at = NULL, + started_initializing_eval_at = NULL, + started_running_eval_at = NULL, + finished_or_errored_at = NULL + WHERE evaluation_run_id = $1 + """, + evaluation_run_id, + EvaluationRunStatus.pending.value, + ) + + logger.info(f"Created attempt {row['attempt_number']} for evaluation run {evaluation_run_id} and reset mirror") + + return _parse_attempt_from_row(row) diff --git a/tests/api/test_retry_policy.py b/tests/api/test_retry_policy.py new file mode 100644 index 00000000..7f491f11 --- /dev/null +++ b/tests/api/test_retry_policy.py @@ -0,0 +1,40 @@ +import pytest + +from models.evaluation_run import EvaluationRunErrorCode, is_retryable_error_code + + +@pytest.mark.parametrize( + "code", + [ + EvaluationRunErrorCode.VALIDATOR_INTERNAL_ERROR, # 2000 + EvaluationRunErrorCode.VALIDATOR_FAILED_PENDING, # 2010 + EvaluationRunErrorCode.VALIDATOR_FAILED_INIT_AGENT, # 2020 + EvaluationRunErrorCode.VALIDATOR_FAILED_RUNNING_AGENT, # 2030 + EvaluationRunErrorCode.VALIDATOR_FAILED_INIT_EVAL, # 2040 + EvaluationRunErrorCode.VALIDATOR_FAILED_RUNNING_EVAL, # 2050 + EvaluationRunErrorCode.PLATFORM_FAILED_PROVISIONING, # 3050 + ], +) +def test_retryable_codes(code): + assert is_retryable_error_code(int(code)) is True + + +@pytest.mark.parametrize( + "code", + [ + int(EvaluationRunErrorCode.AGENT_EXCEPTION_RUNNING_AGENT), # 1000: agent fault + int(EvaluationRunErrorCode.AGENT_INVALID_PATCH), # 1040: agent fault + int(EvaluationRunErrorCode.VALIDATOR_UNKNOWN_PROBLEM), # 2060: deterministic + int(EvaluationRunErrorCode.PLATFORM_RESTARTED_WHILE_PENDING), # 3000: session dead + int(EvaluationRunErrorCode.PLATFORM_RESTARTED_WHILE_RUNNING_EVAL), # 3040: session dead + int(EvaluationRunErrorCode.PLATFORM_PRUNED_BY_SCORE_BOUND), # 3060: pointless + 1234, # unknown 1xxx-range code + 9999, # out of range + ], +) +def test_non_retryable_codes(code): + assert is_retryable_error_code(code) is False + + +def test_none_is_not_retryable(): + assert is_retryable_error_code(None) is False diff --git a/tests/api/test_run_retry_e2e.py b/tests/api/test_run_retry_e2e.py new file mode 100644 index 00000000..535be919 --- /dev/null +++ b/tests/api/test_run_retry_e2e.py @@ -0,0 +1,201 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from uuid import uuid4 + +import pytest + +import utils.database as _db +from api.endpoints import validator as validator_endpoint +from api.endpoints.validator_models import ValidatorUpdateEvaluationRunRequest +from models.evaluation_run import EvaluationRunErrorCode, EvaluationRunStatus +from models.evaluation_set import EvaluationSetGroup, EvaluationSetProblem +from models.problem import ProblemTestCategory, ProblemTestResult, ProblemTestResultStatus +from queries.agent import get_agent_by_id +from queries.evaluation import get_evaluation_by_id, get_hydrated_evaluation_by_id +from queries.evaluation_run import create_evaluation_runs, get_all_evaluation_runs_in_evaluation_id +from queries.evaluation_run_attempt import get_attempts_for_evaluation_run + +pytestmark = pytest.mark.anyio + + +@pytest.fixture(autouse=True) +async def clean_tables(postgres_db): + yield + async with _db.pool.acquire() as conn: + await conn.execute( + "TRUNCATE evaluation_run_logs, evaluation_run_attempts, evaluation_runs, evaluations," + " agents, evaluation_sets RESTART IDENTITY CASCADE" + ) + + +async def _seed(conn): + agent_id = uuid4() + await conn.execute( + "INSERT INTO evaluation_sets (set_id, set_group, problem_name, created_at)" + " VALUES (1, 'validator', 'prob-1', $1)", + datetime(2026, 7, 1, tzinfo=timezone.utc), + ) + await conn.execute( + "INSERT INTO agents (agent_id, miner_hotkey, name, version_num, status, created_at, ip_address)" + " VALUES ($1, '5FakeHotkey', 'agent-a', 1, 'evaluating', NOW(), '127.0.0.1')", + agent_id, + ) + evaluation_id = uuid4() + await conn.execute( + "INSERT INTO evaluations (evaluation_id, agent_id, validator_hotkey, set_id, created_at," + " evaluation_set_group) VALUES ($1, $2, 'validator-hotkey', 1, NOW(), 'validator')", + evaluation_id, + agent_id, + ) + return agent_id, evaluation_id + + +async def _make_session(evaluation_id, agent_id): + evaluation = await get_evaluation_by_id(evaluation_id) + agent = await get_agent_by_id(agent_id) + return validator_endpoint.Validator( + session_id=uuid4(), + name="validator-1", + hotkey="validator-hotkey", + time_connected=datetime.now(timezone.utc), + ip_address="127.0.0.1", + current_evaluation_id=evaluation_id, + current_evaluation=evaluation, + current_agent=agent, + ) + + +async def _update(validator, run_id, status, **extra): + request = ValidatorUpdateEvaluationRunRequest(evaluation_run_id=run_id, updated_status=status, **extra) + return await validator_endpoint.validator_update_evaluation_run.__wrapped__(request, validator=validator) + + +async def test_errored_run_is_retried_and_evaluation_succeeds(monkeypatch, postgres_db): + async def fake_presign(_s3_key): + return "https://s3.example.com/fresh-upload-url" + + monkeypatch.setattr(validator_endpoint, "generate_presigned_upload_url", fake_presign) + + async with _db.pool.acquire() as conn: + agent_id, evaluation_id = await _seed(conn) + + await create_evaluation_runs( + evaluation_id, + [ + EvaluationSetProblem( + set_id=1, + set_group=EvaluationSetGroup.validator, + problem_name="prob-1", + created_at=datetime(2026, 7, 1, tzinfo=timezone.utc), + ) + ], + ) + (run,) = await get_all_evaluation_runs_in_evaluation_id(evaluation_id) + validator = await _make_session(evaluation_id, agent_id) + + # Attempt 1 fails with a retryable validator error. + await _update(validator, run.evaluation_run_id, EvaluationRunStatus.initializing_agent) + response = await _update( + validator, + run.evaluation_run_id, + EvaluationRunStatus.error, + error_code=int(EvaluationRunErrorCode.VALIDATOR_INTERNAL_ERROR), + error_message="transient failure", + agent_logs="attempt 1 agent logs", + eval_logs="attempt 1 eval logs", + ) + assert response.retry is True + assert response.attempt_number == 2 + assert response.artifact_upload_url == "https://s3.example.com/fresh-upload-url" + + hydrated = await get_hydrated_evaluation_by_id(evaluation_id) + assert hydrated.status.value == "running" # mirror reset keeps the evaluation open + + # While the retried run is pending, finish-evaluation must refuse to close the evaluation. + from fastapi import HTTPException + + from api.endpoints.validator_models import ValidatorFinishEvaluationRequest + + with pytest.raises(HTTPException) as exc_info: + await validator_endpoint.validator_finish_evaluation.__wrapped__( + ValidatorFinishEvaluationRequest(), validator=validator + ) + assert exc_info.value.status_code == 409 + + # Attempt 2 walks the full state machine and finishes solved. + await _update(validator, run.evaluation_run_id, EvaluationRunStatus.initializing_agent) + await _update(validator, run.evaluation_run_id, EvaluationRunStatus.running_agent) + await _update( + validator, + run.evaluation_run_id, + EvaluationRunStatus.initializing_eval, + patch="the patch", + agent_logs="attempt 2 agent logs", + ) + await _update(validator, run.evaluation_run_id, EvaluationRunStatus.running_eval) + response = await _update( + validator, + run.evaluation_run_id, + EvaluationRunStatus.finished, + verifier_reward=1.0, + test_results=[ + ProblemTestResult(name="t", category=ProblemTestCategory.default, status=ProblemTestResultStatus.PASS) + ], + eval_logs="attempt 2 eval logs", + ) + assert response.retry is False + + # The eval contract is preserved: the evaluation is a success scored from the final attempt. + hydrated = await get_hydrated_evaluation_by_id(evaluation_id) + assert hydrated.status.value == "success" + assert hydrated.score == 1.0 + + attempts = await get_attempts_for_evaluation_run(run.evaluation_run_id) + assert [a.attempt_number for a in attempts] == [1, 2] + assert attempts[0].status == EvaluationRunStatus.error + assert attempts[1].status == EvaluationRunStatus.finished + + +async def test_retry_denied_once_attempt_cap_reached(monkeypatch, postgres_db): + import api.config as config + + async def fake_presign(_s3_key): + return "https://s3.example.com/fresh-upload-url" + + monkeypatch.setattr(validator_endpoint, "generate_presigned_upload_url", fake_presign) + monkeypatch.setattr(config, "MAX_ATTEMPTS_PER_EVALUATION_RUN", 2) + + async with _db.pool.acquire() as conn: + agent_id, evaluation_id = await _seed(conn) + + await create_evaluation_runs( + evaluation_id, + [ + EvaluationSetProblem( + set_id=1, + set_group=EvaluationSetGroup.validator, + problem_name="prob-1", + created_at=datetime(2026, 7, 1, tzinfo=timezone.utc), + ) + ], + ) + (run,) = await get_all_evaluation_runs_in_evaluation_id(evaluation_id) + validator = await _make_session(evaluation_id, agent_id) + + def _error_kwargs(n): + return dict( + error_code=int(EvaluationRunErrorCode.VALIDATOR_INTERNAL_ERROR), + error_message=f"failure {n}", + agent_logs=f"attempt {n} agent logs", + eval_logs=f"attempt {n} eval logs", + ) + + response = await _update(validator, run.evaluation_run_id, EvaluationRunStatus.error, **_error_kwargs(1)) + assert response.retry is True # attempt 2 granted + + response = await _update(validator, run.evaluation_run_id, EvaluationRunStatus.error, **_error_kwargs(2)) + assert response.retry is False # cap of 2 reached + + hydrated = await get_hydrated_evaluation_by_id(evaluation_id) + assert hydrated.status.value == "failure" # falls back to today's full-restart path diff --git a/tests/api/test_run_retry_grant.py b/tests/api/test_run_retry_grant.py new file mode 100644 index 00000000..aeabd033 --- /dev/null +++ b/tests/api/test_run_retry_grant.py @@ -0,0 +1,208 @@ +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from uuid import uuid4 + +import pytest + +import api.config as config +from api.endpoints import validator as validator_endpoint +from api.endpoints.validator_models import ValidatorUpdateEvaluationRunRequest +from models.agent import Agent, AgentStatus +from models.evaluation import Evaluation +from models.evaluation_run import ( + EvaluationRun, + EvaluationRunAttempt, + EvaluationRunErrorCode, + EvaluationRunStatus, +) +from models.evaluation_set import EvaluationSetGroup + +pytestmark = pytest.mark.anyio + + +def _make_setup(monkeypatch, *, agent_status=AgentStatus.evaluating, attempt_count=1): + evaluation_id = uuid4() + agent_id = uuid4() + + evaluation_run = EvaluationRun( + evaluation_run_id=uuid4(), + evaluation_id=evaluation_id, + problem_name="prob-1", + status=EvaluationRunStatus.running_agent, + created_at=datetime.now(timezone.utc) - timedelta(minutes=5), + started_initializing_agent_at=datetime.now(timezone.utc) - timedelta(minutes=4), + started_running_agent_at=datetime.now(timezone.utc) - timedelta(minutes=3), + ) + + evaluation = Evaluation( + evaluation_id=evaluation_id, + agent_id=agent_id, + validator_hotkey="validator-hotkey", + set_id=1, + evaluation_set_group=EvaluationSetGroup.validator, + created_at=datetime.now(timezone.utc), + ) + + agent = Agent( + agent_id=agent_id, + miner_hotkey="5FakeHotkey", + name="agent-a", + version_num=1, + status=agent_status, + created_at=datetime.now(timezone.utc), + ) + + validator = validator_endpoint.Validator( + session_id=uuid4(), + name="validator-1", + hotkey="validator-hotkey", + time_connected=datetime.now(timezone.utc), + ip_address="127.0.0.1", + current_evaluation_id=evaluation_id, + current_evaluation=evaluation, + current_agent=agent, + ) + + capture = {"created_attempts": 0} + + async def fake_get_evaluation_run_by_id(_run_id): + return evaluation_run + + async def fake_update_evaluation_run_by_id(_run): + return None + + async def fake_create_evaluation_run_log(_run_id, _type, _logs): + return None + + async def fake_check_if_evaluation_run_logs_exist(_run_id, _type): + return True + + async def fake_get_agent_by_id(_agent_id): + return agent + + async def fake_maybe_stop_agent_by_score_bound(_evaluation): + return False + + async def fake_get_attempt_count(_run_id): + return attempt_count + + async def fake_create_next_attempt(run_id): + capture["created_attempts"] += 1 + return EvaluationRunAttempt( + attempt_id=uuid4(), + evaluation_run_id=run_id, + attempt_number=attempt_count + 1, + status=EvaluationRunStatus.pending, + created_at=datetime.now(timezone.utc), + ) + + async def fake_generate_presigned_upload_url(_s3_key): + return "https://s3.example.com/fresh-upload-url" + + monkeypatch.setattr(validator_endpoint, "get_evaluation_run_by_id", fake_get_evaluation_run_by_id) + monkeypatch.setattr(validator_endpoint, "update_evaluation_run_by_id", fake_update_evaluation_run_by_id) + monkeypatch.setattr(validator_endpoint, "create_evaluation_run_log", fake_create_evaluation_run_log) + monkeypatch.setattr( + validator_endpoint, "check_if_evaluation_run_logs_exist", fake_check_if_evaluation_run_logs_exist + ) + monkeypatch.setattr(validator_endpoint, "get_agent_by_id", fake_get_agent_by_id) + monkeypatch.setattr(validator_endpoint, "_maybe_stop_agent_by_score_bound", fake_maybe_stop_agent_by_score_bound) + monkeypatch.setattr(validator_endpoint, "get_attempt_count_for_evaluation_run", fake_get_attempt_count) + monkeypatch.setattr(validator_endpoint, "create_next_attempt_and_reset_evaluation_run", fake_create_next_attempt) + monkeypatch.setattr(validator_endpoint, "generate_presigned_upload_url", fake_generate_presigned_upload_url) + + return evaluation_run, validator, capture + + +def _error_request(evaluation_run, error_code): + return ValidatorUpdateEvaluationRunRequest( + evaluation_run_id=evaluation_run.evaluation_run_id, + updated_status=EvaluationRunStatus.error, + error_code=int(error_code), + error_message="boom", + agent_logs="agent logs", + eval_logs="eval logs", + ) + + +async def _call_update(request, validator): + return await validator_endpoint.validator_update_evaluation_run.__wrapped__(request, validator=validator) + + +async def test_retryable_error_grants_retry(monkeypatch) -> None: + evaluation_run, validator, capture = _make_setup(monkeypatch) + + response = await _call_update( + _error_request(evaluation_run, EvaluationRunErrorCode.VALIDATOR_INTERNAL_ERROR), validator + ) + + assert response.retry is True + assert response.attempt_number == 2 + assert response.artifact_upload_url == "https://s3.example.com/fresh-upload-url" + assert capture["created_attempts"] == 1 + + +async def test_agent_error_is_not_retried(monkeypatch) -> None: + evaluation_run, validator, capture = _make_setup(monkeypatch) + + response = await _call_update(_error_request(evaluation_run, EvaluationRunErrorCode.AGENT_INVALID_PATCH), validator) + + assert response.retry is False + assert capture["created_attempts"] == 0 + + +async def test_attempt_cap_blocks_retry(monkeypatch) -> None: + evaluation_run, validator, capture = _make_setup(monkeypatch, attempt_count=config.MAX_ATTEMPTS_PER_EVALUATION_RUN) + + response = await _call_update( + _error_request(evaluation_run, EvaluationRunErrorCode.VALIDATOR_INTERNAL_ERROR), validator + ) + + assert response.retry is False + assert capture["created_attempts"] == 0 + + +async def test_legacy_run_without_attempts_is_not_retried(monkeypatch) -> None: + evaluation_run, validator, capture = _make_setup(monkeypatch, attempt_count=0) + + response = await _call_update( + _error_request(evaluation_run, EvaluationRunErrorCode.VALIDATOR_INTERNAL_ERROR), validator + ) + + assert response.retry is False + assert capture["created_attempts"] == 0 + + +async def test_stopped_agent_blocks_retry(monkeypatch) -> None: + evaluation_run, validator, capture = _make_setup(monkeypatch, agent_status=AgentStatus.cancelled) + + response = await _call_update( + _error_request(evaluation_run, EvaluationRunErrorCode.VALIDATOR_INTERNAL_ERROR), validator + ) + + assert response.retry is False + assert capture["created_attempts"] == 0 + + +async def test_finished_update_never_retries(monkeypatch) -> None: + from models.problem import ProblemTestCategory, ProblemTestResult, ProblemTestResultStatus + + evaluation_run, validator, capture = _make_setup(monkeypatch) + evaluation_run.patch = "existing patch" + + response = await _call_update( + ValidatorUpdateEvaluationRunRequest( + evaluation_run_id=evaluation_run.evaluation_run_id, + updated_status=EvaluationRunStatus.finished, + verifier_reward=1.0, + test_results=[ + ProblemTestResult(name="t", category=ProblemTestCategory.default, status=ProblemTestResultStatus.PASS) + ], + eval_logs="eval logs", + ), + validator, + ) + + assert response.retry is False + assert capture["created_attempts"] == 0 diff --git a/tests/execution/test_job_names.py b/tests/execution/test_job_names.py new file mode 100644 index 00000000..02079933 --- /dev/null +++ b/tests/execution/test_job_names.py @@ -0,0 +1,14 @@ +from uuid import UUID + +from execution.engine import _format_job_name + +RUN_ID = UUID("12345678-1234-5678-1234-567812345678") + + +def test_attempt_one_keeps_legacy_job_name(): + assert _format_job_name(problem_name="prob", evaluation_run_id=RUN_ID) == f"prob__{RUN_ID}" + assert _format_job_name(problem_name="prob", evaluation_run_id=RUN_ID, attempt_number=1) == f"prob__{RUN_ID}" + + +def test_later_attempts_get_a_suffix(): + assert _format_job_name(problem_name="prob", evaluation_run_id=RUN_ID, attempt_number=2) == f"prob__{RUN_ID}__a2" diff --git a/tests/queries/test_evaluation_run_attempts.py b/tests/queries/test_evaluation_run_attempts.py new file mode 100644 index 00000000..7c5a19d1 --- /dev/null +++ b/tests/queries/test_evaluation_run_attempts.py @@ -0,0 +1,276 @@ +from datetime import datetime, timezone +from uuid import uuid4 + +import pytest + +import utils.database as _db +from models.evaluation_run import EvaluationRunLogType, EvaluationRunStatus +from models.evaluation_set import EvaluationSetGroup, EvaluationSetProblem +from queries.evaluation import update_unfinished_evaluation_runs_in_evaluation_id_to_errored +from queries.evaluation_run import ( + check_if_evaluation_run_logs_exist, + create_evaluation_run_log, + create_evaluation_runs, + get_all_evaluation_runs_in_evaluation_id, + get_evaluation_run_by_id, + get_evaluation_run_logs_by_id, + get_evaluation_run_metrics_by_id, + get_evaluation_run_status_and_attempt_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, + get_attempts_for_evaluation_run, +) + +pytestmark = pytest.mark.anyio + + +@pytest.fixture(autouse=True) +async def clean_tables(postgres_db): + yield + async with _db.pool.acquire() as conn: + await conn.execute( + "TRUNCATE evaluation_run_logs, evaluation_run_attempts, evaluation_runs, evaluations," + " agents, evaluation_sets RESTART IDENTITY CASCADE" + ) + + +async def _seed_evaluation(conn, *, agent_status: str = "evaluating"): + """Insert the minimal agent + evaluation pair that evaluation runs hang off.""" + agent_id = uuid4() + await conn.execute( + "INSERT INTO evaluation_sets (set_id, set_group, problem_name, created_at)" + " VALUES (1, 'validator', 'prob-1', $1) ON CONFLICT DO NOTHING", + datetime(2026, 7, 1, tzinfo=timezone.utc), + ) + await conn.execute( + "INSERT INTO agents (agent_id, miner_hotkey, name, version_num, status, created_at, ip_address)" + " VALUES ($1, '5FakeHotkey', 'agent-a', 1, $2, NOW(), '127.0.0.1')", + agent_id, + agent_status, + ) + evaluation_id = uuid4() + await conn.execute( + "INSERT INTO evaluations (evaluation_id, agent_id, validator_hotkey, set_id, created_at," + " evaluation_set_group) VALUES ($1, $2, 'validator-hotkey', 1, NOW(), 'validator')", + evaluation_id, + agent_id, + ) + return agent_id, evaluation_id + + +def _problem(name: str = "prob-1") -> EvaluationSetProblem: + return EvaluationSetProblem( + set_id=1, + set_group=EvaluationSetGroup.validator, + problem_name=name, + created_at=datetime(2026, 7, 1, tzinfo=timezone.utc), + ) + + +async def test_create_evaluation_runs_creates_attempt_one(postgres_db): + async with _db.pool.acquire() as conn: + _, evaluation_id = await _seed_evaluation(conn) + + await create_evaluation_runs(evaluation_id, [_problem()]) + runs = await get_all_evaluation_runs_in_evaluation_id(evaluation_id) + assert len(runs) == 1 + + attempts = await get_attempts_for_evaluation_run(runs[0].evaluation_run_id) + assert len(attempts) == 1 + assert attempts[0].attempt_number == 1 + assert attempts[0].status == EvaluationRunStatus.pending + assert await get_attempt_count_for_evaluation_run(runs[0].evaluation_run_id) == 1 + + +async def test_attempt_count_is_zero_for_legacy_run(postgres_db): + async with _db.pool.acquire() as conn: + _, evaluation_id = await _seed_evaluation(conn) + legacy_run_id = uuid4() + await conn.execute( + "INSERT INTO evaluation_runs (evaluation_run_id, evaluation_id, problem_name, status, created_at)" + " VALUES ($1, $2, 'prob-1', 'pending', NOW())", + legacy_run_id, + evaluation_id, + ) + + assert await get_attempt_count_for_evaluation_run(legacy_run_id) == 0 + assert await get_attempts_for_evaluation_run(legacy_run_id) == [] + + +async def _seed_single_run(conn): + _, evaluation_id = await _seed_evaluation(conn) + await create_evaluation_runs(evaluation_id, [_problem()]) + runs = await get_all_evaluation_runs_in_evaluation_id(evaluation_id) + return evaluation_id, runs[0] + + +async def test_update_evaluation_run_dual_writes_current_attempt(postgres_db): + async with _db.pool.acquire() as conn: + _, run = await _seed_single_run(conn) + + run.status = EvaluationRunStatus.error + run.error_code = 2000 + run.error_message = "boom" + run.cost_usd = 1.25 + run.finished_or_errored_at = datetime.now(timezone.utc) + await update_evaluation_run_by_id(run) + + (attempt,) = await get_attempts_for_evaluation_run(run.evaluation_run_id) + assert attempt.status == EvaluationRunStatus.error + assert attempt.error_code == 2000 + assert attempt.error_message == "boom" + assert attempt.cost_usd == 1.25 + assert attempt.finished_or_errored_at is not None + + +async def test_create_next_attempt_resets_mirror(postgres_db): + async with _db.pool.acquire() as conn: + _, run = await _seed_single_run(conn) + + run.status = EvaluationRunStatus.error + run.error_code = 2000 + run.error_message = "boom" + run.patch = "some patch" + run.finished_or_errored_at = datetime.now(timezone.utc) + await update_evaluation_run_by_id(run) + + new_attempt = await create_next_attempt_and_reset_evaluation_run(run.evaluation_run_id) + assert new_attempt.attempt_number == 2 + assert new_attempt.status == EvaluationRunStatus.pending + + mirrored = await get_evaluation_run_by_id(run.evaluation_run_id) + assert mirrored.status == EvaluationRunStatus.pending + assert mirrored.error_code is None + assert mirrored.error_message is None + assert mirrored.patch is None + assert mirrored.cost_usd is None + assert mirrored.finished_or_errored_at is None + assert mirrored.started_initializing_agent_at is None + + attempts = await get_attempts_for_evaluation_run(run.evaluation_run_id) + assert [a.attempt_number for a in attempts] == [1, 2] + assert attempts[0].status == EvaluationRunStatus.error # history preserved + + +async def test_dual_write_targets_latest_attempt_only(postgres_db): + async with _db.pool.acquire() as conn: + _, run = await _seed_single_run(conn) + + run.status = EvaluationRunStatus.error + run.error_code = 2000 + run.error_message = "boom" + run.finished_or_errored_at = datetime.now(timezone.utc) + await update_evaluation_run_by_id(run) + await create_next_attempt_and_reset_evaluation_run(run.evaluation_run_id) + + refreshed = await get_evaluation_run_by_id(run.evaluation_run_id) + refreshed.status = EvaluationRunStatus.initializing_agent + refreshed.started_initializing_agent_at = datetime.now(timezone.utc) + await update_evaluation_run_by_id(refreshed) + + attempts = await get_attempts_for_evaluation_run(run.evaluation_run_id) + assert attempts[0].status == EvaluationRunStatus.error # untouched + assert attempts[1].status == EvaluationRunStatus.initializing_agent + + +async def test_bulk_error_update_mirrors_current_attempt(postgres_db): + async with _db.pool.acquire() as conn: + evaluation_id, run = await _seed_single_run(conn) + + await update_unfinished_evaluation_runs_in_evaluation_id_to_errored(evaluation_id, "validator died") + + mirrored = await get_evaluation_run_by_id(run.evaluation_run_id) + assert mirrored.status == EvaluationRunStatus.error + + (attempt,) = await get_attempts_for_evaluation_run(run.evaluation_run_id) + assert attempt.status == EvaluationRunStatus.error + assert attempt.error_code == 3000 # PLATFORM_RESTARTED_WHILE_PENDING + assert attempt.error_message == "validator died" + assert attempt.finished_or_errored_at is not None + + +async def test_logs_are_attempt_scoped(postgres_db): + async with _db.pool.acquire() as conn: + _, run = await _seed_single_run(conn) + + await create_evaluation_run_log(run.evaluation_run_id, EvaluationRunLogType.agent, "attempt 1 logs") + assert await check_if_evaluation_run_logs_exist(run.evaluation_run_id, EvaluationRunLogType.agent) is True + + run.status = EvaluationRunStatus.error + run.error_code = 2000 + run.error_message = "boom" + run.finished_or_errored_at = datetime.now(timezone.utc) + await update_evaluation_run_by_id(run) + await create_next_attempt_and_reset_evaluation_run(run.evaluation_run_id) + + # A fresh attempt has no logs yet, so the insert-once guard must allow a new insert. + assert await check_if_evaluation_run_logs_exist(run.evaluation_run_id, EvaluationRunLogType.agent) is False + await create_evaluation_run_log(run.evaluation_run_id, EvaluationRunLogType.agent, "attempt 2 logs") + + # Reads return the latest attempt's logs; attempt 1 logs remain in the table. + assert await get_evaluation_run_logs_by_id(run.evaluation_run_id, EvaluationRunLogType.agent) == "attempt 2 logs" + async with _db.pool.acquire() as conn: + count = await conn.fetchval( + "SELECT COUNT(*) FROM evaluation_run_logs WHERE evaluation_run_id = $1", run.evaluation_run_id + ) + assert count == 2 + + +async def test_metrics_include_attempt_count(postgres_db): + async with _db.pool.acquire() as conn: + _, run = await _seed_single_run(conn) + + metrics = await get_evaluation_run_metrics_by_id(run.evaluation_run_id) + assert metrics["attempt_count"] == 1 + + run.status = EvaluationRunStatus.error + run.error_code = 2000 + run.error_message = "boom" + run.finished_or_errored_at = datetime.now(timezone.utc) + await update_evaluation_run_by_id(run) + await create_next_attempt_and_reset_evaluation_run(run.evaluation_run_id) + + metrics = await get_evaluation_run_metrics_by_id(run.evaluation_run_id) + assert metrics["attempt_count"] == 2 + + +async def test_metrics_attempt_count_defaults_to_one_for_legacy_run(postgres_db): + async with _db.pool.acquire() as conn: + _, evaluation_id = await _seed_evaluation(conn) + legacy_run_id = uuid4() + await conn.execute( + "INSERT INTO evaluation_runs (evaluation_run_id, evaluation_id, problem_name, status, created_at)" + " VALUES ($1, $2, 'prob-1', 'pending', NOW())", + legacy_run_id, + evaluation_id, + ) + + metrics = await get_evaluation_run_metrics_by_id(legacy_run_id) + assert metrics["attempt_count"] == 1 + + +async def test_status_and_attempt_lookup(postgres_db): + async with _db.pool.acquire() as conn: + _, run = await _seed_single_run(conn) + + assert await get_evaluation_run_status_and_attempt_by_id(run.evaluation_run_id) == ( + EvaluationRunStatus.pending, + 1, + ) + + run.status = EvaluationRunStatus.error + run.error_code = 2000 + run.error_message = "boom" + run.finished_or_errored_at = datetime.now(timezone.utc) + await update_evaluation_run_by_id(run) + await create_next_attempt_and_reset_evaluation_run(run.evaluation_run_id) + + assert await get_evaluation_run_status_and_attempt_by_id(run.evaluation_run_id) == ( + EvaluationRunStatus.pending, + 2, + ) + + assert await get_evaluation_run_status_and_attempt_by_id(uuid4()) is None diff --git a/tests/test_cost_hash_map.py b/tests/test_cost_hash_map.py new file mode 100644 index 00000000..52ee32df --- /dev/null +++ b/tests/test_cost_hash_map.py @@ -0,0 +1,17 @@ +from uuid import uuid4 + +from inference_gateway.cost_hash_map import CostHashMap + + +def test_attempts_have_independent_budgets(): + cost_map = CostHashMap() + run_id = uuid4() + + cost_map.add_cost((run_id, 1), 1.5) + assert cost_map.get_cost((run_id, 1)) == 1.5 + # A new attempt starts from a zero budget. + assert cost_map.get_cost((run_id, 2)) == 0 + + cost_map.add_cost((run_id, 2), 0.25) + assert cost_map.get_cost((run_id, 2)) == 0.25 + assert cost_map.get_cost((run_id, 1)) == 1.5 diff --git a/tests/validator/__init__.py b/tests/validator/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/validator/test_main_status_flow.py b/tests/validator/test_main_status_flow.py index 1b149c0d..b2d63868 100644 --- a/tests/validator/test_main_status_flow.py +++ b/tests/validator/test_main_status_flow.py @@ -7,7 +7,7 @@ import pytest import validator.main as validator_main -from api.endpoints.validator_models import ValidatorUpdateEvaluationRunRequest +from api.endpoints.validator_models import ValidatorUpdateEvaluationRunRequest, ValidatorUpdateEvaluationRunResponse from execution.errors import EvaluationRunException from execution.types import ExecutionResult, TrialSnapshot from models.evaluation_run import EvaluationRunErrorCode, EvaluationRunStatus @@ -64,6 +64,7 @@ async def fake_update(evaluation_run_id, problem_name, updated_status, extra=Non "problem_name": problem_name, } ) + return ValidatorUpdateEvaluationRunResponse() monkeypatch.setattr(validator_main, "update_evaluation_run", fake_update) return capture diff --git a/tests/validator/test_retry_loop.py b/tests/validator/test_retry_loop.py new file mode 100644 index 00000000..5aea5044 --- /dev/null +++ b/tests/validator/test_retry_loop.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +from types import SimpleNamespace +from uuid import uuid4 + +import pytest + +import validator.main as validator_main +from api.endpoints.validator_models import ( + ValidatorRequestEvaluationResponseEvaluationRun, + ValidatorUpdateEvaluationRunResponse, +) +from execution.errors import EvaluationRunException +from models.evaluation_run import EvaluationRunErrorCode, EvaluationRunStatus + +pytestmark = pytest.mark.anyio + + +def _fake_result(job_dir=None): + return SimpleNamespace( + backend="harbor", + job_dir=job_dir, + patch="a patch", + agent_logs="agent logs", + eval_logs="eval logs", + verifier_reward=1.0, + test_results=[], + cost_usd=0.1, + ) + + +class FakeEngine: + """Fails the first attempt with a retryable validator error, succeeds on the second.""" + + def __init__(self): + self.calls: list[int] = [] + + async def evaluate(self, **kwargs): + self.calls.append(kwargs.get("attempt_number", 1)) + if len(self.calls) == 1: + raise EvaluationRunException(EvaluationRunErrorCode.VALIDATOR_INTERNAL_ERROR, "transient failure") + return _fake_result() + + +@pytest.fixture +def run_stub(): + return ValidatorRequestEvaluationResponseEvaluationRun( + evaluation_run_id=uuid4(), problem_name="prob-1", execution_spec=None + ) + + +async def test_retry_directive_reruns_the_attempt(monkeypatch, run_stub): + engine = FakeEngine() + monkeypatch.setattr(validator_main, "execution_engine", engine) + monkeypatch.setattr(validator_main, "max_evaluation_run_log_size_bytes", 1_000_000) + + updates: list[EvaluationRunStatus] = [] + + async def fake_update_evaluation_run(evaluation_run_id, problem_name, updated_status, extra=None, *, timeout=None): + updates.append(updated_status) + if updated_status == EvaluationRunStatus.error: + return ValidatorUpdateEvaluationRunResponse( + retry=True, attempt_number=2, artifact_upload_url="https://s3.example.com/fresh" + ) + return ValidatorUpdateEvaluationRunResponse() + + monkeypatch.setattr(validator_main, "update_evaluation_run", fake_update_evaluation_run) + + await validator_main._run_evaluation_run(run_stub, "agent code") + + assert engine.calls == [1, 2] # second attempt carried attempt_number=2 + assert updates.count(EvaluationRunStatus.error) == 1 + assert updates.count(EvaluationRunStatus.finished) == 1 + # Two attempts each start by moving pending -> initializing_agent. + assert updates.count(EvaluationRunStatus.initializing_agent) == 2 + + +async def test_denied_retry_stops_after_one_attempt(monkeypatch, run_stub): + class AlwaysFailingEngine: + def __init__(self): + self.calls = 0 + + async def evaluate(self, **kwargs): + self.calls += 1 + raise EvaluationRunException(EvaluationRunErrorCode.VALIDATOR_INTERNAL_ERROR, "persistent failure") + + engine = AlwaysFailingEngine() + monkeypatch.setattr(validator_main, "execution_engine", engine) + monkeypatch.setattr(validator_main, "max_evaluation_run_log_size_bytes", 1_000_000) + + async def fake_update_evaluation_run(evaluation_run_id, problem_name, updated_status, extra=None, *, timeout=None): + return ValidatorUpdateEvaluationRunResponse() # retry always False + + monkeypatch.setattr(validator_main, "update_evaluation_run", fake_update_evaluation_run) + + await validator_main._run_evaluation_run(run_stub, "agent code") + + assert engine.calls == 1 diff --git a/validator/main.py b/validator/main.py index 8c66d586..aba7bdcb 100644 --- a/validator/main.py +++ b/validator/main.py @@ -27,6 +27,7 @@ ValidatorRequestEvaluationResponse, ValidatorTaskDownloadUrlRequest, ValidatorUpdateEvaluationRunRequest, + ValidatorUpdateEvaluationRunResponse, ) from execution.artifacts import _read_proxy_cost from execution.engine import ExecutionEngine @@ -87,7 +88,7 @@ async def update_evaluation_run( extra: Dict[str, Any] | None = None, *, timeout: int | None = None, -): +) -> ValidatorUpdateEvaluationRunResponse: logger.info(f"Updating evaluation run {evaluation_run_id} for problem {problem_name} to {updated_status.value}...") clean_extra = {k: v for k, v in (extra or {}).items() if v is not None} @@ -101,11 +102,12 @@ async def update_evaluation_run( request = ValidatorUpdateEvaluationRunRequest( evaluation_run_id=evaluation_run_id, updated_status=updated_status, **clean_extra ) - await retry_with_backoff( + response_data = await retry_with_backoff( lambda: post_ridges_platform("/validator/update-evaluation-run", request, **post_kwargs), max_attempts=3, base_delay=2.0, ) + return ValidatorUpdateEvaluationRunResponse(**(response_data or {})) # Truncates a log if required @@ -237,149 +239,183 @@ async def _run_evaluation_run_with_semaphore( ) -# Run an evaluation run -async def _run_evaluation_run( +async def _run_single_attempt( evaluation_run, agent_code: str, - artifact_upload_url: str | None = None, - openrouter_config: OpenRouterRuntimeConfig | None = None, -): + attempt_number: int, + artifact_upload_url: str | None, + openrouter_config: OpenRouterRuntimeConfig | None, +) -> ValidatorUpdateEvaluationRunResponse: + """Run one attempt of an evaluation run and report its terminal status. + + Returns the platform's response to the terminal update; response.retry + signals that a fresh attempt should be started. + """ + global execution_engine + assert execution_engine is not None + evaluation_run_id = evaluation_run.evaluation_run_id problem_name = evaluation_run.problem_name - task_digest = (evaluation_run.execution_spec or {}).get("task_digest") - if task_digest: - _active_task_digests.add(task_digest) + job_dir = None + terminal_response = ValidatorUpdateEvaluationRunResponse() try: - global execution_engine - assert execution_engine is not None - - logger.info(f"Starting evaluation run {evaluation_run_id} for problem {problem_name}...") - - job_dir = None - - try: - # Move from pending -> initializing_agent - await update_evaluation_run(evaluation_run_id, problem_name, EvaluationRunStatus.initializing_agent) - - async def _on_agent_started() -> None: - await update_evaluation_run( - evaluation_run_id, - problem_name, - EvaluationRunStatus.running_agent, - timeout=STATUS_HOOK_TIMEOUT_SECONDS, - ) - - async def _on_verification_started(snapshot: TrialSnapshot) -> None: - await update_evaluation_run( - evaluation_run_id, - problem_name, - EvaluationRunStatus.initializing_eval, - { - "patch": snapshot.patch, - "agent_logs": truncate_logs_if_required(snapshot.agent_logs), - }, - timeout=STATUS_HOOK_TIMEOUT_SECONDS, - ) - - await update_evaluation_run( - evaluation_run_id, - problem_name, - EvaluationRunStatus.running_eval, - timeout=STATUS_HOOK_TIMEOUT_SECONDS, - ) - - result = await execution_engine.evaluate( - evaluation_run_id=evaluation_run_id, - problem_name=problem_name, - execution_spec=evaluation_run.execution_spec, - agent_path=None, - agent_code=agent_code, - openrouter_config=openrouter_config, - fetch_task_url=_fetch_task_download_url, - on_agent_started=_on_agent_started, - on_verification_started=_on_verification_started, - ) - job_dir = result.job_dir + # Move from pending -> initializing_agent + await update_evaluation_run(evaluation_run_id, problem_name, EvaluationRunStatus.initializing_agent) - logger.info( - f"Finished {result.backend} execution for problem {problem_name}: " - f"{len(result.patch.splitlines())} lines of patch, " - f"{len(result.agent_logs.splitlines())} lines of agent logs, " - f"{len(result.eval_logs.splitlines())} lines of eval logs" - ) - - num_passed = sum(1 for test in result.test_results if test.status == ProblemTestResultStatus.PASS) - num_failed = sum(1 for test in result.test_results if test.status == ProblemTestResultStatus.FAIL) - num_skipped = sum(1 for test in result.test_results if test.status == ProblemTestResultStatus.SKIP) - logger.info( - f"Finished running evaluation for problem {problem_name}: " - f"reward={result.verifier_reward}, {len(result.test_results)} test results " - f"({num_passed} passed, {num_failed} failed, {num_skipped} skipped), " - f"{len(result.eval_logs.splitlines())} lines of eval logs" - ) - - # Move from running_eval -> finished + async def _on_agent_started() -> None: await update_evaluation_run( evaluation_run_id, problem_name, - EvaluationRunStatus.finished, - { - "patch": result.patch, - "agent_logs": truncate_logs_if_required(result.agent_logs), - "verifier_reward": result.verifier_reward, - "test_results": [ - test.model_dump(exclude={"test_alias"}, exclude_none=True) for test in result.test_results - ], - "eval_logs": truncate_logs_if_required(result.eval_logs), - "cost_usd": result.cost_usd, - }, + EvaluationRunStatus.running_agent, + timeout=STATUS_HOOK_TIMEOUT_SECONDS, ) - except EvaluationRunException as e: - logger.error(f"Evaluation run {evaluation_run_id} for problem {problem_name} errored: {e}") - extra = dict(e.extra or {}) - job_dir = extra.pop("job_dir", None) - for key in ("agent_logs", "eval_logs"): - if key in extra: - extra[key] = truncate_logs_if_required(extra[key]) - - cost_usd = _read_proxy_cost(job_dir) if job_dir else None - + async def _on_verification_started(snapshot: TrialSnapshot) -> None: await update_evaluation_run( evaluation_run_id, problem_name, - EvaluationRunStatus.error, + EvaluationRunStatus.initializing_eval, { - "error_code": e.error_code.value, - "error_message": e.error_message, - "cost_usd": cost_usd, - **extra, + "patch": snapshot.patch, + "agent_logs": truncate_logs_if_required(snapshot.agent_logs), }, + timeout=STATUS_HOOK_TIMEOUT_SECONDS, ) - except Exception as e: - logger.error( - f"Evaluation run {evaluation_run_id} for problem {problem_name} errored: {EvaluationRunErrorCode.VALIDATOR_INTERNAL_ERROR.get_error_message()}: {e}" - ) - logger.error(traceback.format_exc()) - await update_evaluation_run( evaluation_run_id, problem_name, - EvaluationRunStatus.error, - { - "error_code": EvaluationRunErrorCode.VALIDATOR_INTERNAL_ERROR.value, - "error_message": ( - f"{EvaluationRunErrorCode.VALIDATOR_INTERNAL_ERROR.get_error_message()}: {e}\n\n" - f"Traceback:\n{traceback.format_exc()}" - ), - }, + EvaluationRunStatus.running_eval, + timeout=STATUS_HOOK_TIMEOUT_SECONDS, ) - # Upload artifacts for both success and error cases - if artifact_upload_url and job_dir: - await _upload_job_artifacts(job_dir, artifact_upload_url) + result = await execution_engine.evaluate( + evaluation_run_id=evaluation_run_id, + problem_name=problem_name, + execution_spec=evaluation_run.execution_spec, + agent_path=None, + agent_code=agent_code, + openrouter_config=openrouter_config, + fetch_task_url=_fetch_task_download_url, + on_agent_started=_on_agent_started, + on_verification_started=_on_verification_started, + attempt_number=attempt_number, + ) + job_dir = result.job_dir + + logger.info( + f"Finished {result.backend} execution for problem {problem_name}: " + f"{len(result.patch.splitlines())} lines of patch, " + f"{len(result.agent_logs.splitlines())} lines of agent logs, " + f"{len(result.eval_logs.splitlines())} lines of eval logs" + ) + + num_passed = sum(1 for test in result.test_results if test.status == ProblemTestResultStatus.PASS) + num_failed = sum(1 for test in result.test_results if test.status == ProblemTestResultStatus.FAIL) + num_skipped = sum(1 for test in result.test_results if test.status == ProblemTestResultStatus.SKIP) + logger.info( + f"Finished running evaluation for problem {problem_name}: " + f"reward={result.verifier_reward}, {len(result.test_results)} test results " + f"({num_passed} passed, {num_failed} failed, {num_skipped} skipped), " + f"{len(result.eval_logs.splitlines())} lines of eval logs" + ) + + # Move from running_eval -> finished + terminal_response = await update_evaluation_run( + evaluation_run_id, + problem_name, + EvaluationRunStatus.finished, + { + "patch": result.patch, + "agent_logs": truncate_logs_if_required(result.agent_logs), + "verifier_reward": result.verifier_reward, + "test_results": [ + test.model_dump(exclude={"test_alias"}, exclude_none=True) for test in result.test_results + ], + "eval_logs": truncate_logs_if_required(result.eval_logs), + "cost_usd": result.cost_usd, + }, + ) + + except EvaluationRunException as e: + logger.error(f"Evaluation run {evaluation_run_id} for problem {problem_name} errored: {e}") + extra = dict(e.extra or {}) + job_dir = extra.pop("job_dir", None) + for key in ("agent_logs", "eval_logs"): + if key in extra: + extra[key] = truncate_logs_if_required(extra[key]) + + cost_usd = _read_proxy_cost(job_dir) if job_dir else None + + terminal_response = await update_evaluation_run( + evaluation_run_id, + problem_name, + EvaluationRunStatus.error, + { + "error_code": e.error_code.value, + "error_message": e.error_message, + "cost_usd": cost_usd, + **extra, + }, + ) + + except Exception as e: + logger.error( + f"Evaluation run {evaluation_run_id} for problem {problem_name} errored: {EvaluationRunErrorCode.VALIDATOR_INTERNAL_ERROR.get_error_message()}: {e}" + ) + logger.error(traceback.format_exc()) + + terminal_response = await update_evaluation_run( + evaluation_run_id, + problem_name, + EvaluationRunStatus.error, + { + "error_code": EvaluationRunErrorCode.VALIDATOR_INTERNAL_ERROR.value, + "error_message": ( + f"{EvaluationRunErrorCode.VALIDATOR_INTERNAL_ERROR.get_error_message()}: {e}\n\n" + f"Traceback:\n{traceback.format_exc()}" + ), + }, + ) + + # Upload artifacts for both success and error cases (per attempt; the run's + # S3 key always holds the latest attempt's artifacts) + if artifact_upload_url and job_dir: + await _upload_job_artifacts(job_dir, artifact_upload_url) + + return terminal_response + + +# Run an evaluation run (all of its attempts) +async def _run_evaluation_run( + evaluation_run, + agent_code: str, + artifact_upload_url: str | None = None, + openrouter_config: OpenRouterRuntimeConfig | None = None, +): + evaluation_run_id = evaluation_run.evaluation_run_id + problem_name = evaluation_run.problem_name + task_digest = (evaluation_run.execution_spec or {}).get("task_digest") + if task_digest: + _active_task_digests.add(task_digest) + + try: + logger.info(f"Starting evaluation run {evaluation_run_id} for problem {problem_name}...") + + attempt_number = 1 + while True: + response = await _run_single_attempt( + evaluation_run, agent_code, attempt_number, artifact_upload_url, openrouter_config + ) + if not response.retry: + break + attempt_number = response.attempt_number or attempt_number + 1 + artifact_upload_url = response.artifact_upload_url or artifact_upload_url + logger.info( + f"Platform granted a retry for evaluation run {evaluation_run_id} " + f"(problem {problem_name}); starting attempt {attempt_number}" + ) logger.info(f"Finished evaluation run {evaluation_run_id} for problem {problem_name}")