fix(agent-challenge): unblock Terminal-Bench scoring and surface trial failures

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

@@ -40,7 +40,9 @@
 import contextlib
 import importlib
 import inspect
-from collections.abc import Awaitable, Callable
+import os
+import tempfile
+from collections.abc import Awaitable, Callable, Iterator
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
@@ -73,6 +75,92 @@
 #: agent-timeout code -- kept distinct from per-command crashes.
 AGENT_TIMEOUT_REASON_CODE = "harbor_agent_timeout_error"
 
+#: Maximum characters of captured construction output appended to an error.
+CAPTURED_OUTPUT_LIMIT = 600
+
+
+@contextlib.contextmanager
+def _capture_process_output() -> Iterator[Callable[[], str]]:
+    """Capture fd 1/2 for the duration of the block, then replay them.
+
+    A miner constructor typically shells out (``pip install ...``). The child
+    writes to the *inherited* file descriptors, so Python-level redirection
+    never sees it and ``CalledProcessError`` keeps only argv + exit status —
+    the actual reason ("Directory is not installable", a resolver conflict, a
+    missing wheel) is lost from the trial record. Capturing at the fd level is
+    the only way to keep it.
+
+    Each stream is replayed to its original descriptor on exit so the runner's
+    own logs stay byte-complete; the returned callable yields the captured text
+    and is only meaningful after the block has exited.
+    """
+
+    captured: dict[str, str] = {"text": ""}
+
+    def read() -> str:
+        return captured["text"]
+
+    saved: dict[int, int] = {}
+    buffers: dict[int, Any] = {}
+    try:
+        for fd in (1, 2):
+            handle = tempfile.TemporaryFile()
+            saved[fd] = os.dup(fd)
+            os.dup2(handle.fileno(), fd)
+            buffers[fd] = handle
+    except OSError:
+        # Never let observability break execution: restore and run uncaptured.
+        for fd, original in saved.items():
+            with contextlib.suppress(OSError):
+                os.dup2(original, fd)
+                os.close(original)
+        for handle in buffers.values():
+            with contextlib.suppress(OSError):
+                handle.close()
+        yield read
+        return
+
+    try:
+        yield read
+    finally:
+        chunks: list[str] = []
+        for fd in (1, 2):
+            data = b""
+            handle = buffers.get(fd)
+            if handle is not None:
+                try:
+                    handle.flush()
+                    handle.seek(0)
+                    data = handle.read()
+                except OSError:
+                    data = b""
+            original = saved.get(fd)
+            if original is not None:
+                with contextlib.suppress(OSError):
+                    os.dup2(original, fd)
+                    os.close(original)
+            if data:
+                with contextlib.suppress(OSError):
+                    os.write(fd, data)
+                chunks.append(data.decode("utf-8", "replace"))
+            if handle is not None:
+                with contextlib.suppress(OSError):
+                    handle.close()
+        captured["text"] = "".join(chunks)
+
+
+def _captured_detail(text: str) -> str:
+    """Render captured output as a single-line suffix for an error message."""
+
+    lines = [line.strip() for line in text.splitlines() if line.strip()]
+    if not lines:
+        return ""
+    joined = " / ".join(lines)
+    if len(joined) > CAPTURED_OUTPUT_LIMIT:
+        joined = joined[-CAPTURED_OUTPUT_LIMIT:]
+        joined = f"…{joined}"
+    return f" | output: {joined}"
+
 
 class AgentLoadError(Exception):
     """The agent class could not be located / loaded (typed, fail-fast).
@@ -246,24 +334,41 @@ async def drive(
         session: TmuxSession | None = None
         try:
             # 1. Load + construct the agent. Any failure here is the
-            #    submission's fault -> submission_code_failed.
-            try:
-                agent_cls = self._agent_class or load_agent_class(self._import_path)
-                init_kwargs = dict(self._extra_init_kwargs)
-                if resolved_agent_env:
-                    init_kwargs.setdefault("extra_env", dict(resolved_agent_env))
-                agent = agent_cls(
-                    logs_dir=Path(logs_dir) if logs_dir is not None else None,
-                    model_name=model_name,
-                    **init_kwargs,
+            #    submission's fault -> submission_code_failed. Construction is
+            #    wrapped in an fd-level capture because miner constructors shell
+            #    out, and the child's diagnostics would otherwise never reach
+            #    the trial record.
+            load_error: AgentLoadError | None = None
+            construction_error: Exception | None = None
+            with _capture_process_output() as captured_output:
+                try:
+                    agent_cls = self._agent_class or load_agent_class(self._import_path)
+                    init_kwargs = dict(self._extra_init_kwargs)
+                    if resolved_agent_env:
+                        init_kwargs.setdefault("extra_env", dict(resolved_agent_env))
+                    agent = agent_cls(
+                        logs_dir=Path(logs_dir) if logs_dir is not None else None,
+                        model_name=model_name,
+                        **init_kwargs,
+                    )
+                except AgentLoadError as exc:
+                    load_error = exc
+                except Exception as exc:
+                    construction_error = exc
+            if load_error is not None:
+                return AgentRunResult(
+                    status="failed",
+                    reason_code=load_error.reason_code,
+                    error=f"{load_error}{_captured_detail(captured_output())}",
                 )
-            except AgentLoadError as exc:
-                return AgentRunResult(status="failed", reason_code=exc.reason_code, error=str(exc))
-            except Exception as exc:
+            if construction_error is not None:
                 return AgentRunResult(
                     status="failed",
                     reason_code=AGENT_LOAD_FAILED_REASON_CODE,
-                    error=f"agent construction failed: {exc}",
+                    error=(
+                        f"agent construction failed: {construction_error}"
+                        f"{_captured_detail(captured_output())}"
+                    ),
                 )
 
             agent_info = self._safe_agent_info(agent)

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: