fix(agent-challenge): stop a job once the agent cannot be constructed

Agent construction -- unpack, install, import, instantiate -- happens per
trial but does not depend on the task. A submission whose ZIP ships a bare
agent.py with no installable project fails identically on every task, so a
30-task job spent hours of wall clock and real LLM budget re-deriving one
packaging error, 30 times, before reporting 0.

Count construction failures across trials and short-circuit the remaining
ones once a small threshold confirms the package is broken. The threshold is
greater than one so a single transient fault, such as a flaky package index
during pip install, cannot abort an otherwise healthy job. Trials that never
ran are still planned, still aggregated, and still score 0, so the totals
stay honest; each carries the construction reason code and an error text
naming the short-circuit and what to fix.

Author: echobt

packages/challenges/agent-challenge/src/agent_challenge/evaluation/own_runner/orchestrator.py

@@ -43,7 +43,10 @@
 from pathlib import Path
 from typing import Any
 
-from agent_challenge.evaluation.own_runner.driver import AgentDriver
+from agent_challenge.evaluation.own_runner.driver import (
+    AGENT_LOAD_FAILED_REASON_CODE,
+    AgentDriver,
+)
 from agent_challenge.evaluation.own_runner.reason_codes import (
     REASON_CODES,
     is_known_reason_code,
@@ -105,6 +108,17 @@
 #: finalizes with its sibling trials intact.
 TRIAL_CRASH_REASON_CODE = "harbor_trial_failed"
 
+#: How many construction failures confirm a broken submission package.
+#:
+#: Agent construction (unpack + install + import + instantiate) is
+#: task-independent: a ZIP missing its manifest, or importing a module it never
+#: shipped, breaks identically on every task. Re-running all 30 tasks to
+#: relearn that one fact burns hours of wall clock and real LLM budget, so the
+#: job stops once this many trials have failed that way. The threshold is >1 so
+#: a single transient fault (a flaky package index during ``pip install``)
+#: cannot abort an otherwise healthy job.
+CONSTRUCTION_FAILURE_ABORT_THRESHOLD = 2
+
 # Fail fast at import if the taxonomy ever drops a code we emit.
 assert TRIAL_TIMEOUT_REASON_CODE in REASON_CODES
 assert TRIAL_CRASH_REASON_CODE in REASON_CODES
@@ -443,21 +457,36 @@ async def run(self, tasks: Sequence[TaskSpec]) -> JobResult:
         state_lock = asyncio.Lock()
         in_flight = 0
         peak = 0
+        construction_failures = 0
+        short_circuit = False
 
         async def execute(trial_id: TrialId) -> TrialOutcome:
-            nonlocal in_flight, peak
+            nonlocal in_flight, peak, construction_failures, short_circuit
             # Resume: a persisted result means this trial is already done -- load
             # it WITHOUT acquiring the semaphore (it never re-runs, never counts
             # toward in-flight, never double-counts).
             persisted = self._load_trial(trial_id)
             if persisted is not None:
                 return persisted
 
+            task = task_lookup[trial_id.task_name]
+
+            # Fail-fast: once enough trials have proven the submission's agent
+            # cannot be constructed, the remaining trials would burn budget to
+            # reproduce the same packaging error. Resolve them immediately as
+            # explicit, self-describing failures instead of running them.
+            async with state_lock:
+                aborted = short_circuit
+            if aborted:
+                outcome = self._short_circuited_outcome(trial_id, task)
+                self._persist_trial(trial_id, outcome)
+                await self._notify_trial_listener(trial_id, outcome)
+                return outcome
+
             async with semaphore:
                 async with state_lock:
                     in_flight += 1
                     peak = max(peak, in_flight)
-                task = task_lookup[trial_id.task_name]
                 try:
                     # Backstop: bound the whole trial (prepare + drive + verify +
                     # teardown) so one stalled sub-step can never wedge
@@ -481,6 +510,13 @@ async def execute(trial_id: TrialId) -> TrialOutcome:
                 finally:
                     async with state_lock:
                         in_flight -= 1
+                # Count construction failures so a broken package trips the
+                # fail-fast gate above for every trial not yet started.
+                if outcome.reason_code == AGENT_LOAD_FAILED_REASON_CODE:
+                    async with state_lock:
+                        construction_failures += 1
+                        if construction_failures >= CONSTRUCTION_FAILURE_ABORT_THRESHOLD:
+                            short_circuit = True
                 # Persist immediately so a later crash cannot lose a finished
                 # trial (and a resume skips it).
                 self._persist_trial(trial_id, outcome)
@@ -541,6 +577,34 @@ def _crashed_outcome(
             error_text=f"trial crashed: {type(exc).__name__}: {exc}",
         )
 
+    def _short_circuited_outcome(self, trial_id: TrialId, task: TaskSpec) -> TrialOutcome:
+        """Failed outcome for a trial never run because the package is broken.
+
+        Mirrors :meth:`_timed_out_outcome` (``status="failed"``, ``errored=True``,
+        ``rewards=None``) so the job still aggregates every planned trial and the
+        totals stay honest -- the task was planned, scored 0, and says exactly
+        why it never executed.
+        """
+
+        return TrialOutcome(
+            task_name=trial_id.task_name,
+            trial_name=trial_id.trial_name,
+            status="failed",
+            rewards=None,
+            reason_code=AGENT_LOAD_FAILED_REASON_CODE,
+            errored=True,
+            agent_name=self._config.agent_name,
+            model_name=self._config.model_name,
+            source=task.source,
+            error_text=(
+                "short-circuit: agent construction failed on "
+                f"{CONSTRUCTION_FAILURE_ABORT_THRESHOLD} earlier trials, so this "
+                "trial was not run. Agent construction is task-independent -- fix "
+                "the submission package (installable project + importable modules) "
+                "and resubmit."
+            ),
+        )
+
     # -- lock / persistence ------------------------------------------------
 
     def _check_or_write_lock(self) -> None:

packages/challenges/agent-challenge/tests/test_own_runner_orchestrator.py

@@ -963,3 +963,49 @@ async def _preparer(trial_id: TrialId, task: TaskSpec) -> PreparedTrial:
     assert result.resolved == 2
     assert result.total == 2
     assert result.pass_at_k == {"agent__adhoc": {2: 1.0}}
+
+
+# ===========================================================================
+# Fail-fast: a submission whose agent cannot even be constructed
+# ===========================================================================
+
+
+async def test_construction_failures_short_circuit_the_remaining_trials(
+    tmp_path: Path,
+) -> None:
+    """A package that cannot construct fails identically on every task.
+
+    Agent construction is task-independent: a ZIP missing its manifest or its
+    imported modules breaks the same way for all 30 tasks. Running them all
+    burns hours of LLM budget to relearn one fact, so the job must stop after a
+    small confirmation threshold and mark the rest explicitly.
+    """
+    attempted: list[str] = []
+
+    async def _run(trial_id: TrialId, task: TaskSpec) -> TrialOutcome:
+        attempted.append(trial_id.trial_name)
+        return TrialOutcome(
+            task_name=trial_id.task_name,
+            trial_name=trial_id.trial_name,
+            status="failed",
+            rewards=None,
+            reason_code="harbor_submission_code_failed",
+            errored=True,
+            error_text="agent construction failed: no pyproject.toml",
+        )
+
+    tasks = [TaskSpec(f"task-{i}") for i in range(10)]
+    orch = TrialJobOrchestrator(
+        config=JobConfig(n_attempts=1, n_concurrent=1),
+        job_dir=tmp_path / "job",
+        trial_runner=_run,
+    )
+    result = await orch.run(tasks)
+
+    assert len(attempted) <= 2, f"must stop early, ran {len(attempted)} trials"
+    # Every task still accounted for, and the skipped ones say why.
+    assert result.n_total_trials == 10
+    assert result.score == 0.0
+    skipped = [o for o in result.trial_outcomes if "short-circuit" in (o.error_text or "")]
+    assert len(skipped) == 10 - len(attempted)
+    assert all(o.reason_code == "harbor_submission_code_failed" for o in skipped)