default/claude_code

"""Claude Code — a single-agent baseline that delegates the entire search loop to the Claude Code CLI.

Galapagos variant of SkyDiscover's ``claude_code`` baseline controller with two deliberate changes:
the CLI runs **locally on the host** (SkyDiscover wraps it in a Docker runner image) and billing is
**subscription-only** — ``ANTHROPIC_API_KEY`` / ``ANTHROPIC_AUTH_TOKEN`` are always stripped from
the CLI's environment so a run can never silently bill the API. Auth comes from
``CLAUDE_CODE_OAUTH_TOKEN`` (``claude setup-token``, Harbor-style) or, as a fallback, a copy of the
local ``claude /login`` credentials.

The scaffold stages a workspace (the seed as ``solution<suffix>``, a ``run_eval.sh`` scoring
command, a ``TASK.md`` brief), launches one ``claude -p`` session with
``--max-turns`` set to a safety cap, and lets Claude iterate on the solution file against the
evaluator on its own. While the session runs, the framework polls the solution file and scores each
new version with the task's Evaluator, so the Population / trajectory / checkpoints record real
intermediate progress; the final solution is scored the same way when the session ends.

Trace capture follows Harbor's design: ``CLAUDE_CONFIG_DIR`` is redirected to
``<out_dir>/claude_sessions``, so the CLI writes its **native session JSONL**
(``claude_sessions/projects/**/<session>.jsonl``) directly into the run's output directory —
byte-identical, incrementally, with no copy step, surviving any crash. That native trace is the
session's source-of-truth record. The raw stream-json stdout is still teed to ``claude.log``
(it carries what the native trace lacks: ``total_cost_usd`` and the live turn accounting), and
``run_summary.json`` records the session totals.

SkyDiscover requires Docker because ``--dangerously-skip-permissions`` needs isolation; here that
flag is passed to a process running as *your user on your machine* — the agent has full local
access for the duration of the run. Only point it at tasks whose evaluators you trust.
"""
from __future__ import annotations

import json
import logging
import os
import shlex
import shutil
import signal
import subprocess
import sys
import tempfile
import threading
import time
import uuid
from pathlib import Path

from ...components import (DefaultPromptBuilder, ExploreExploitPolicy, LLMProposer,
                           InMemoryPopulation, NullMemory)
from ...components.evaluator import _RUNNER, ContainerEvaluator, _default_hf_cache_runtime
from ...environments import (EVAL_DIR, TaskContainer, build_agent_image, dood, provision_task_dir,
                             resolve_image)
from ...config import GalapagosConfig
from ...models import GalapagosModel
from ...records import Genome
from ..._logging import STATUS, fmt_new_best_callout, fmt_new_best_marker
from ..base_scaffold import GalapagosScaffold, _event_line, _invalid_reason
from ..registry import register_scaffold

log = logging.getLogger(__name__)

def _describe_turn(content_blocks: list, limit: int = 88) -> str:
    """What the agent did on this turn, in one line.

    The old rendering printed ``Assistant response`` for every turn that was not a tool call, so a
    session read as eleven identical lines and told the operator nothing about *why* the run was
    taking ten minutes. A turn is worth exactly one line, and that line should name the action: the
    tool plus its target (which command, which file), or the head of the assistant's own text. That
    is the difference between "the agent is alive" and "the agent is recompiling for the fifth time".
    """
    def _clip(s: str) -> str:
        s = " ".join(str(s).split())
        return s[:limit] + ("…" if len(s) > limit else "")

    tools, text = [], ""
    for c in content_blocks:
        if not isinstance(c, dict):
            continue
        if c.get("type") == "tool_use":
            name = c.get("name", "tool")
            inp = c.get("input") if isinstance(c.get("input"), dict) else {}
            # the argument that identifies the action, in the order a reader would want it
            target = (inp.get("command") or inp.get("file_path") or inp.get("pattern")
                      or inp.get("description") or "")
            tools.append(f"{name}: {_clip(target)}" if target else name)
        elif c.get("type") == "text" and not text:
            text = str(c.get("text") or "")
    if tools:
        return "  ".join(tools)
    if text.strip():
        return _clip(text)
    return "thinking…"   # a turn of pure reasoning: say so rather than call it a "response"


_CLAUDE_MODEL_PREFIXES = ("claude-", "sonnet", "opus", "haiku")
_EFFORT_LEVELS = ("low", "medium", "high", "xhigh", "max")   # `claude --effort` accepted values
_POLL_INTERVAL_S = 10      # how often the solution file is checked for a new version to score
_HARD_STOP_GRACE_S = 30    # after the stream exceeds the turn budget, wait this long for a result
_DEFAULT_TURN_OVERHEAD_S = 300  # per CLI-capped turn: thinking, editing, shell overhead
_MIN_WALL_TIMEOUT_S = 1800      # never derive less than 30 minutes by default

# Tools the benchmark agent must not have.
#
# The first three are merely useless in a headless ``-p`` session (they wait for a human).
#
# WebSearch/WebFetch are a different matter: they let the agent look up the ANSWER. This is not
# hypothetical — in the recorded traces under outputs/, the agent WebSearched a FrontierCS problem's
# published editorial ("题解"), and WebFetched github.com/google-deepmind/alphaevolve_results asking
# whether it held "solutions or notebooks" for the very AlphaEvolve problem it was being scored on.
# Our own task prompts hand it the search key: 447 of the 1,235 cards name the source benchmark's
# arXiv id in `system_message`.
#
# It also breaks the leaderboard's premise. A `search`-tier scaffold (a frozen LLM behind a diff
# operator) *cannot* browse; claude_code could. Comparing them on one task was never apples-to-apples.
#
# This has to be a DENYlist, not an allowlist: the session runs under
# ``--dangerously-skip-permissions``, which bypasses permission checks, so ``--allowedTools`` — a
# permission concept — does not constrain anything, while ``--disallowedTools`` still removes the tool.
# Verified by A/B against the real CLI: with the old list the model calls WebSearch; with this one it
# reports the tool is unavailable and calls nothing.
#
# NOT closed by this: the agent still has Bash on a networked container and could `curl` out. Nothing
# in the traces does (every curl goes to the task's own judge on localhost), and closing it for real
# means egress filtering on the agent container — the CLI itself needs to reach the Anthropic API, so
# `--network none` is not available. Tracked separately; this shuts the door that was actually used.
_DISALLOWED_TOOLS = "AskUserQuestion,EnterPlanMode,ExitPlanMode,WebSearch,WebFetch"

# The redirected CLAUDE_CONFIG_DIR (Harbor-style) under the run's output directory. The CLI's
# native session JSONL lands at <out_dir>/claude_sessions/projects/**/<session>.jsonl — the
# byte-identical source-of-truth trace, written incrementally by the CLI itself.
_CLAUDE_SESSIONS_DIRNAME = "claude_sessions"

# Subdirectories Claude Code expects inside its config dir (mirrors Harbor's setup command).
_CLAUDE_CONFIG_SUBDIRS = ("projects", "todos", "debug", "shell-snapshots", "statsig", "skills")


def _prepare_claude_config_dir(out_dir: Path) -> Path:
    """Create ``<out_dir>/claude_sessions`` with the skeleton the CLI expects and return it."""
    config_dir = out_dir / _CLAUDE_SESSIONS_DIRNAME
    for sub in _CLAUDE_CONFIG_SUBDIRS:
        (config_dir / sub).mkdir(parents=True, exist_ok=True)
    return config_dir


def _setup_subscription_auth(env: dict, claude_config_dir: Path) -> str:
    """Force subscription-only billing; returns the auth source used.

    The CLI silently prefers an API key over the subscription whenever one is visible, so both
    key variables are unconditionally dropped — a run can never bill the API. Auth then comes
    from, in order:

    1. ``CLAUDE_CODE_OAUTH_TOKEN`` (a long-lived token from ``claude setup-token``; Harbor-style,
       works with a redirected ``CLAUDE_CONFIG_DIR`` with no extra state), or
    2. a copy of ``~/.claude/.credentials.json`` (the interactive ``/login`` state) placed into
       the redirected config dir, since the CLI only looks for it there.
    """
    for var in ("ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN"):
        env.pop(var, None)
    if (env.get("CLAUDE_CODE_OAUTH_TOKEN") or "").strip():
        return "oauth-token"
    # Whole-loop container orchestration cannot rely on the container's HOME carrying the host login.
    # It bind-mounts that one file read-only and names it here; host-mode callers keep the normal path.
    credentials = Path(env.get("GALAPAGOS_CLAUDE_CREDENTIALS_FILE")
                       or (Path.home() / ".claude" / ".credentials.json"))
    if credentials.exists():
        target = claude_config_dir / ".credentials.json"
        shutil.copy(credentials, target)
        target.chmod(0o600)
        log.info("using ~/.claude/.credentials.json (copied into the run's CLAUDE_CONFIG_DIR); "
                 "prefer CLAUDE_CODE_OAUTH_TOKEN (`claude setup-token`) for headless runs")
        return "login-credentials"
    raise RuntimeError(
        "claude_code is subscription-only and no subscription auth was found: set "
        "CLAUDE_CODE_OAUTH_TOKEN (run `claude setup-token` once) or sign in with `claude` + "
        "/login (creates ~/.claude/.credentials.json). API-key billing is not supported.")


def _native_session_files(claude_config_dir: Path) -> list[str]:
    """Relative paths (under out_dir) of every native session JSONL the CLI wrote."""
    projects = claude_config_dir / "projects"
    if not projects.is_dir():
        return []
    return sorted(str(path.relative_to(claude_config_dir.parent))
                  for path in projects.rglob("*.jsonl") if path.is_file())


def _result_error(evt: dict) -> str:
    """One-line description of a CLI ``result`` event that carries ``is_error``.

    ``is_error`` is overloaded and on its own means nothing: an ``error_max_turns`` result — the
    session simply spending its turn budget — sets it too, and that is a *normal* end to a run that
    did all its work. So the flag is only ever read alongside "did the session produce anything"
    (see the guard in the run loops); this just names the failure for the message.
    """
    status = evt.get("api_error_status")
    detail = str(evt.get("result") or evt.get("subtype") or "unknown error").strip()
    return f"API error {status}: {detail}" if status else detail


def _assert_session_did_something(stats: dict, ckpt_count: int, returncode: int | None,
                                  log_path: Path) -> None:
    """Fail loudly when the CLI session ended in an error having produced nothing whatsoever.

    A session that dies before it can work — an expired subscription token 401ing on the first call
    is the canonical case — otherwise passes for a *legitimate* run: no candidate is ever admitted,
    so the Population still holds only the seed, the run "completes", and the sweep records the
    SEED's score as the agent's result. That is worse than a zero, because a zero is visible. An
    auth outage instead launders a plausible-looking number onto the leaderboard, and a whole sweep
    can rot this way without a single line of red.

    The discriminator is deliberately NOT ``is_error``: an ``error_max_turns`` result sets it too,
    and spending the turn budget is simply how a healthy run ends. It is *did the session produce
    anything* — no candidate written AND nothing billed. A session that got a checkpoint scored, or
    that burned so much as a token, did real work; whatever killed it afterwards (a 529, the wall
    timeout) still leaves a partial result worth keeping.
    """
    if ckpt_count or stats["cost_usd"]:
        return
    failure = stats["error"] or (f"the CLI exited with code {returncode}" if returncode else None)
    if failure is None:
        return
    try:
        tail = log_path.read_text(errors="replace")[-2000:]
    except OSError:
        tail = ""
    raise RuntimeError(
        f"the Claude Code session produced nothing — no candidate written, nothing billed "
        f"({failure}). Is subscription auth valid? (CLAUDE_CODE_OAUTH_TOKEN via "
        f"`claude setup-token`, or `claude` + /login). Log tail:\n{tail}")


def _kill_proc_group(proc) -> None:
    """Kill the CLI and everything it spawned (Bash tool children inherit the stdout pipe, so a
    bare ``proc.kill()`` can leave orphans holding the pipe open and hang the reader)."""
    try:
        os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
    except (ProcessLookupError, PermissionError, OSError):
        try:
            proc.kill()
        except OSError:
            pass


# The agent's workspace is bind-mounted here AND is the CLI's HOME, so Claude Code writes its native
# session trace to the DEFAULT `$HOME/.claude/projects/**.jsonl` — no galapagos-specific
# `CLAUDE_CONFIG_DIR`. Generic, reference-agent-style paths (`/workspace`, `$HOME/.claude`) keep the
# recorded trajectory reusable as training data. `/workspace` (not `/run`, a real FHS dir in
# Debian-derived images) is the conventional agent workdir.
WORKSPACE_DIR = "/workspace"
_AGENT_PID_FILE = f"{WORKSPACE_DIR}/.agent.pid"


def _setup_subscription_auth_docker(claude_config_dir: Path) -> tuple[str, dict[str, str]]:
    """Resolve subscription-only auth and return ``(source, env vars to pass with -e)``.

    Unlike the host scaffold, there is no environment to strip: a container gets none of the host's env
    unless explicitly forwarded, so ``ANTHROPIC_API_KEY``/``ANTHROPIC_AUTH_TOKEN`` are never passed in
    the first place.
    """
    token = (os.environ.get("CLAUDE_CODE_OAUTH_TOKEN") or "").strip()
    if token:
        return "oauth-token", {"CLAUDE_CODE_OAUTH_TOKEN": token}
    credentials = Path.home() / ".claude" / ".credentials.json"
    if credentials.exists():
        target = claude_config_dir / ".credentials.json"
        shutil.copy(credentials, target)
        target.chmod(0o600)
        log.info("using ~/.claude/.credentials.json (copied into the run's mounted, redirected "
                 "config dir); prefer CLAUDE_CODE_OAUTH_TOKEN (`claude setup-token`) for headless runs")
        return "login-credentials", {}
    raise RuntimeError(
        "claude_code is subscription-only and no subscription auth was found: set "
        "CLAUDE_CODE_OAUTH_TOKEN (run `claude setup-token` once) or sign in with `claude` + "
        "/login (creates ~/.claude/.credentials.json). API-key billing is not supported.")

@register_scaffold("claude_code")
class ClaudeCodeScaffold(GalapagosScaffold):
    name = "claude_code"
    needs_model = False    # the CLI is both model and proposer; no GalapagosModel is built or called
    manages_own_container = True   # in docker mode it starts the task container itself (agent inside it),
                                   # so the generic CLI orchestrator leaves it alone
    status_unit = "turn"   # the budget buys CLI turns, not loop iterations — a turn may be a tool call
                           # that never touches the program, so calling these "iters" would overstate
                           # how much evolution the run actually bought (see the smoke run: 11 turns → 1 candidate)
    agent_metadata_kind_key = "claude_code"
    agent_metadata_position_key = "claude_code_turn"

    @classmethod
    def build_components(cls, config: GalapagosConfig, model: GalapagosModel | None) -> dict:
        # Claude Code subsumes selection, prompting, and proposal inside one CLI session, so only
        # the Population does real work (it archives the scored checkpoints); the other slots are
        # inert defaults that the overridden _loop never calls.
        return {
            "population": InMemoryPopulation(capacity=None, drop_invalid=True),
            "selection_policy": ExploreExploitPolicy(seed=int(config.seed)),
            "prompt_builder": DefaultPromptBuilder(),
            "proposer": LLMProposer(),
            "memory": NullMemory(),
        }

    # ---- the loop (one CLI session instead of select→propose→evaluate) --------
    def _loop_host(self, task) -> None:
        cfg = self.config.proposer
        prompt_turn_budget = int(self.general.max_iterations)
        if prompt_turn_budget <= 0:
            raise ValueError("claude_code requires general.max_iterations to be positive")
        cli_max_turns = (
            int(cfg.claude_cli_max_turns)
            if cfg.claude_cli_max_turns is not None
            else prompt_turn_budget * 2
        )
        if cli_max_turns < prompt_turn_budget:
            raise ValueError(
                "proposer.claude_cli_max_turns must be >= general.max_iterations "
                f"(got {cli_max_turns} < {prompt_turn_budget})"
            )
        model = cfg.claude_model
        if model and not any(model.startswith(p) for p in _CLAUDE_MODEL_PREFIXES):
            raise ValueError(
                f"claude_code only drives Claude models, got {model!r}; set proposer.claude_model "
                "to a claude-*/sonnet/opus/haiku name (or null for the CLI's default)")
        effort = cfg.reasoning_effort
        if effort is not None and effort not in _EFFORT_LEVELS:
            raise ValueError(
                f"claude_code: invalid proposer.reasoning_effort {effort!r} for `claude --effort`; "
                f"use one of {', '.join(_EFFORT_LEVELS)} (or null for the CLI's default)")
        cli = shutil.which("claude")
        if cli is None:
            raise RuntimeError(
                "the Claude Code CLI ('claude') was not found on PATH — install it "
                "(npm install -g @anthropic-ai/claude-code)")

        seed = self.ctx.best                 # set by setup(): the evaluated seed program
        ev_path = getattr(self.evaluator, "evaluator_path", None)
        if not ev_path:
            raise RuntimeError(f"task {task.name!r} exposes no evaluator path for run_eval.sh")
        eval_timeout = int(getattr(self.evaluator, "timeout", 120))
        suffix = getattr(self.evaluator, "suffix", ".py")

        run_dir = Path(self.ctx.run_dir) if self.ctx.run_dir else None
        if run_dir:
            workspace, cleanup = run_dir / "claude_workspace", False
            workspace.mkdir(parents=True, exist_ok=True)
        else:
            workspace, cleanup = Path(tempfile.mkdtemp(prefix="galapagos-claude-")), True
        out_dir = run_dir or workspace
        log_path, progress_path = out_dir / "claude.log", out_dir / "progress.log"

        # Harbor-style native trace capture + subscription-only auth: the CLI writes its native
        # session JSONL straight into <out_dir>/claude_sessions (no copy step, crash-safe), and
        # the API key variables are stripped so only the claude.ai subscription can be billed.
        claude_config_dir = _prepare_claude_config_dir(out_dir)
        env = dict(os.environ)
        env["CLAUDE_CONFIG_DIR"] = str(claude_config_dir)
        auth_source = _setup_subscription_auth(env, claude_config_dir)

        solution_path = workspace / f"solution{suffix}"
        solution_path.write_text(seed.content if seed else "")
        self._write_eval_script_host(workspace, str(ev_path), eval_timeout)
        prompt_path = workspace / ".prompt.txt"
        prompt_path.write_text(
            self._write_task_prompt_host(workspace, solution_path, prompt_turn_budget, eval_timeout)
        )

        cmd = [cli, "-p", "-", "--max-turns", str(cli_max_turns),
               "--dangerously-skip-permissions", "--disallowedTools", _DISALLOWED_TOOLS,
               "--output-format", "stream-json", "--verbose"]
        if model:
            cmd += ["--model", model]
        if effort:
            cmd += ["--effort", effort]
        if cfg.claude_wall_timeout_seconds is not None:
            wall_timeout = int(cfg.claude_wall_timeout_seconds)
            if wall_timeout <= 0:
                raise ValueError("proposer.claude_wall_timeout_seconds must be positive when set")
        else:
            # Wall-clock safety net: full eval timeout plus generous thinking/editing/shell overhead
            # per CLI-capped turn. This is the whole-session timeout, not a per-turn timeout.
            wall_timeout = max(
                cli_max_turns * (_DEFAULT_TURN_OVERHEAD_S + eval_timeout),
                _MIN_WALL_TIMEOUT_S,
            )

        plock = threading.Lock()

        def progress(line: str) -> None:
            log.info("%s", line)
            with plock, open(progress_path, "a") as f:
                f.write(f"[{time.strftime('%H:%M:%S')}] {line}\n")

        progress(f"Claude Code run started — model={model or 'CLI default'}, "
                 f"effort={effort or 'CLI default'}, "
                 f"prompt_turn_budget={prompt_turn_budget}, cli_max_turns={cli_max_turns}, "
                 f"billing=subscription ({auth_source}), "
                 f"native traces → {claude_config_dir / 'projects'}, "
                 f"wall_timeout={wall_timeout}s")

        stats = {"stream_turns": 0, "cum_turns": 0, "cost_usd": 0.0, "error": None,
                 "session_id": None, "timed_out": False, "returncode": None}
        last_content = seed.content if seed else ""
        ckpt_count = 0
        run_start = time.monotonic()
        proc = None
        try:
            with open(prompt_path) as stdin_fh, open(log_path, "w") as log_fh:
                # start_new_session=True puts the CLI in its own process group so the turn-budget /
                # timeout kill also reaps its tool children (see _kill_proc_group).
                proc = subprocess.Popen(cmd, stdin=stdin_fh, stdout=subprocess.PIPE,
                                        stderr=log_fh, cwd=str(workspace), env=env,
                                        start_new_session=True)
                reader = threading.Thread(
                    target=self._pump_events,
                    args=(proc, log_fh, stats, cli_max_turns, wall_timeout, progress),
                    kwargs={"kill_fn": lambda: _kill_proc_group(proc)},
                    daemon=True)
                reader.start()
                # Poll the solution file while the session runs; score every new version so the
                # run has real checkpoints even if the session dies mid-way.
                while reader.is_alive():
                    reader.join(timeout=_POLL_INTERVAL_S)
                    try:
                        cur = solution_path.read_text()
                    except OSError:
                        continue
                    if cur == last_content or not cur.strip():
                        continue
                    last_content = cur
                    ckpt_count += 1
                    observed_turns = stats["cum_turns"] or stats["stream_turns"]
                    self._admit(cur, seed, iteration=max(observed_turns, ckpt_count),
                                kind="checkpoint", turn=observed_turns)
                    self._maybe_checkpoint()

            actual_turns = stats["cum_turns"] or stats["stream_turns"]
            _assert_session_did_something(stats, ckpt_count, proc.returncode, log_path)

            try:
                final = solution_path.read_text()
            except OSError:
                final = ""
            if final.strip() and final != last_content:   # written between the last poll and exit
                ckpt_count += 1
                self._admit(final, seed, iteration=max(actual_turns, ckpt_count),
                            kind="final", turn=actual_turns)
            self.ctx.iteration = max(self.ctx.iteration, actual_turns, 1)
            if stats["cost_usd"]:
                self.ctx.record_cost(stats["cost_usd"])

            best = self.ctx.best
            summary = {
                "agent_session_format": "claude-native",
                "model": model,
                "effort": effort,
                "billing": "subscription",
                "auth_source": auth_source,
                "session_id": stats["session_id"],
                "native_session_files": _native_session_files(claude_config_dir),
                "prompt_turn_budget": prompt_turn_budget,
                "cli_max_turns": cli_max_turns,
                "actual_turns": actual_turns,
                "checkpoints_scored": ckpt_count,
                "timed_out": stats["timed_out"],
                "return_code": stats["returncode"],
                "cost_usd": round(stats["cost_usd"], 4),
                "wall_seconds": round(time.monotonic() - run_start, 1),
                "baseline_score": seed.scores.get("combined_score") if seed else None,
                "final_score": best.scores.get("combined_score") if best else None,
            }
            (out_dir / "run_summary.json").write_text(json.dumps(summary, indent=2, default=str) + "\n")
            progress(f"Run complete: turns={actual_turns}/{prompt_turn_budget} prompt budget "
                     f"({cli_max_turns} CLI cap), "
                     f"cost=${stats['cost_usd']:.4f}, checkpoints={ckpt_count}, "
                     f"score={summary['final_score']}")
        finally:
            if proc is not None and proc.poll() is None:
                _kill_proc_group(proc)
                proc.wait()
            # The native session JSONL is the session's trace of record — the CLI already wrote
            # it incrementally under claude_sessions/, so there is nothing to persist here.
            native_sessions = _native_session_files(claude_config_dir)
            if native_sessions:
                progress(f"Native Claude Code trace(s) preserved: {', '.join(native_sessions)}")
            else:
                progress("Warning: no native Claude Code session file found under "
                         f"{claude_config_dir / 'projects'}")
            if cleanup:
                shutil.rmtree(workspace, ignore_errors=True)

    # ---- helpers ---------------------------------------------------------------
    def _admit(self, content: str, seed: Genome | None, *, iteration: int, kind: str,
               turn: int) -> None:
        """Score one solution snapshot with the task's Evaluator and add it to the Population,
        updating best/trajectory exactly like the base loop's admission block."""
        parent = getattr(self, "_last_agent_snapshot", None) or seed
        child = Genome(
            content=content,
            parent_id=parent.id if parent else None,
            metadata={
                self.agent_metadata_kind_key: kind,
                self.agent_metadata_position_key: turn,
                "generation": (
                    int(parent.metadata.get("generation", 0)) + 1 if parent else 0
                ),
            },
        )
        _t = time.monotonic()
        try:
            res = self._evaluate_search_candidate(child)
        except Exception as exc:
            self._eval_s = time.monotonic() - _t
            self._eval_total += self._eval_s
            child.trace["evaluation_duration_seconds"] = self._eval_s
            self._record(
                child,
                None,
                origin="agent",
                admitted=None,
                reason="evaluation_failed",
                eval_source="failed",
                evaluation_error=exc,
            )
            raise
        self._eval_s = time.monotonic() - _t
        self._eval_total += self._eval_s
        child.trace["evaluation_duration_seconds"] = self._eval_s
        child.scores = res.metrics
        child.artifacts.update(res.artifacts)
        child.metadata["valid"] = res.valid
        self._consume_population_mutations()
        population_before = {genome.id for genome in self.population.all()}
        self._admission_attempts += 1
        try:
            admitted = self._add_population_candidate(child, result=res)
        except Exception as exc:
            self._record(
                child,
                res,
                origin="agent",
                admitted=None,
                reason="population_update_failed",
                admission_error=exc,
            )
            raise
        population_after = {genome.id for genome in self.population.all()}
        population_removed_ids = sorted(population_before - population_after)
        population_collection_mutations = self._consume_population_mutations()
        self._history.append(child)
        if admitted:
            self._admitted += 1
            self._note_recent(child)
        prev_best = self.ctx.best.fitness if self.ctx.best else float("-inf")
        improved = admitted and child.fitness > prev_best
        if improved:
            self.ctx.best = child
        self.ctx.iteration = max(self.ctx.iteration, iteration)
        best_fit = self.ctx.best.fitness if self.ctx.best else float("-inf")
        if improved and prev_best not in (float("-inf"),):
            delta = child.fitness - prev_best
            marker = fmt_new_best_marker(delta)
        elif improved:
            delta = None
            marker = fmt_new_best_marker()
        else:
            delta = None
            marker = ""
        STATUS.update(iteration=turn, best_score=best_fit,
                      best_id=self.ctx.best.id if self.ctx.best else "",
                      pop=len(self.population), evals=self._evaluation_count(), invalid=self._invalid,
                      cost_usd=self.ctx.cost_usd, score=child.fitness, new_best=improved)
        # The agent's own model calls are billed to a subscription and never touch our LLM path, so
        # there is no `llm_s` to report here — only the framework's scorer is ours to time.
        log.info("%s", _event_line(f"~{turn}", child, res, best_fit, marker, unit=self.status_unit,
                                   eval_s=self._eval_s, pop=len(self.population),
                                   stale=self._stale, cost_usd=self.ctx.cost_usd))
        if improved:
            log.info("  %s", fmt_new_best_callout(child.id, child.fitness, delta))
        if not res.valid:
            reason = _invalid_reason(res)
            if reason:
                log.info("  %s", reason)
        rejection_reason = (
            None if admitted else ("eval_failed" if not res.valid else "not_admitted")
        )
        if rejection_reason:
            self._rejected[rejection_reason] = self._rejected.get(rejection_reason, 0) + 1
        self._record(
            child,
            res,
            origin="agent",
            new_best=improved,
            admitted=admitted,
            reason=rejection_reason,
            metadata={"population_collection_mutations": population_collection_mutations}
            if population_collection_mutations else None,
            population_removed_ids=population_removed_ids,
        )
        self._last_agent_snapshot = child

    @staticmethod
    def _write_eval_script_host(workspace: Path, ev_path: str, timeout: int) -> None:
        """``run_eval.sh`` — the command Claude Code calls to score a candidate. It runs the task's
        own ``evaluator.py`` with THIS interpreter, cwd-anchored to the task dir exactly like
        SubprocessEvaluator, and prints the raw metrics JSON."""
        runner = (
            "import importlib.util, json, sys\n"
            f"spec = importlib.util.spec_from_file_location('_evaluator', {ev_path!r})\n"
            "mod = importlib.util.module_from_spec(spec)\n"
            "spec.loader.exec_module(mod)\n"
            "print(json.dumps(mod.evaluate(sys.argv[1]), default=str))\n"
        )
        (workspace / "_run_eval.py").write_text(runner)
        script = (
            "#!/bin/bash\n"
            "set -euo pipefail\n"
            'PROGRAM="$(realpath "$1")"\n'
            f"cd {shlex.quote(os.path.dirname(os.path.abspath(ev_path)))}\n"
            f'timeout {timeout} {shlex.quote(sys.executable)} '
            f'{shlex.quote(str(workspace / "_run_eval.py"))} "$PROGRAM"\n'
        )
        path = workspace / "run_eval.sh"
        path.write_text(script)
        path.chmod(0o755)

    def _write_task_prompt_host(self, workspace: Path, solution_path: Path, max_turns: int,
                           eval_timeout: int) -> str:
        """``TASK.md`` — the one-shot brief piped to the CLI (SkyDiscover's prompt, with local
        absolute paths instead of ``/workspace``)."""
        content = (
            "You are an AI assistant iteratively improving a program to maximize "
            f"its evaluation score. You have **{max_turns} turns** total.\n\n"
            "## Current solution\n\n"
            f"`{solution_path}` -- read it, understand it, modify it freely.\n\n"
            "## How to evaluate\n\n"
            "```bash\n"
            f"bash {workspace / 'run_eval.sh'} {solution_path}\n"
            "```\n\n"
            "Output is JSON. The `combined_score` field is what you want to maximize "
            f"(higher is better). The evaluator has a **{eval_timeout}s timeout**.\n\n"
            "## Task description\n\n"
            f"{self.ctx.task_context}\n\n"
            "## Instructions\n\n"
            "- Run the evaluator once to confirm the baseline score, then start improving.\n"
            "- After each change, evaluate and decide whether to keep or revert.\n"
            f"- Always keep `{solution_path}` set to your best solution.\n"
            "- Aim to try several distinct approaches within your turn budget.\n"
            f"- Create scratch files only inside `{workspace}`.\n"
        )
        (workspace / "TASK.md").write_text(content)
        return content

    @staticmethod
    def _pump_events(proc, log_fh, stats: dict, cli_max_turns: int, wall_timeout: int,
                     progress, kill_fn=None) -> None:
        """Stream the CLI's stream-json stdout: tee every line to claude.log and track progress.

        The final turn count comes from Claude Code's result.num_turns. While the session is
        running, use distinct assistant message ids as the best available live approximation; this
        counts text-only assistant responses as well as tool-use responses and avoids double-counting
        split/replayed assistant events with the same message id.

        ``kill_fn`` is how a hard stop / wall timeout actually terminates the running session —
        ``_kill_proc_group(proc)`` for a local host process, or a Docker ``kill``/``stop`` call for a
        containerized session (container mode). Defaults to the local process-group kill.
        """
        kill = kill_fn or (lambda: _kill_proc_group(proc))
        start = time.monotonic()
        hard_stop_at = 0.0
        seen_assistant_message_ids: set[str] = set()
        pending: dict | None = None   # the turn whose events are still arriving (see the loop below)

        def _flush_turn() -> None:
            """Print the held-open turn's one line, if it has not been printed already."""
            nonlocal pending
            if not pending:
                return
            progress(f"turn ~{pending['turn']}/{cli_max_turns}  "
                     f"{_describe_turn(pending['blocks'])}  [{time.monotonic() - start:.0f}s]")
            pending = None

        def record(raw: bytes) -> dict | None:
            """Tee one stdout line to claude.log; returns the parsed event dict or None."""
            decoded = raw.decode("utf-8", errors="replace")
            log_fh.write(decoded)
            log_fh.flush()
            try:
                evt = json.loads(raw)
            except (json.JSONDecodeError, ValueError):
                return None
            if stats["session_id"] is None and isinstance(evt, dict) and evt.get("session_id"):
                stats["session_id"] = evt["session_id"]
            return evt if isinstance(evt, dict) else None

        try:
            for raw in proc.stdout:
                evt = record(raw)
                if evt is None:
                    continue
                etype = evt.get("type")
                if etype == "assistant":
                    message = evt.get("message")
                    message = message if isinstance(message, dict) else {}
                    message_id = message.get("id")
                    should_count = True
                    if isinstance(message_id, str) and message_id:
                        if message_id in seen_assistant_message_ids:
                            should_count = False
                        else:
                            seen_assistant_message_ids.add(message_id)
                    if should_count:
                        stats["stream_turns"] += 1
                        STATUS.update(iteration=stats["stream_turns"], cost_usd=stats["cost_usd"])
                        if stats["stream_turns"] > cli_max_turns and not hard_stop_at:
                            hard_stop_at = time.monotonic()
                            progress(f"Hard stop: stream turn {stats['stream_turns']} exceeded "
                                     f"CLI cap {cli_max_turns} — waiting for result")
                    # The CLI streams ONE assistant message as SEVERAL events sharing a message id:
                    # the thinking blocks arrive first and the tool_use last. Counting the turn on
                    # first sight is right (that is when the turn begins), but *describing* it from
                    # that first event is not — it is almost always a bare thinking block, which is
                    # why every turn used to log as "Assistant response" while the tool call it
                    # actually made was dropped on the floor. So describe from the whole message:
                    # hold the turn open, and print its line on the first event that carries an
                    # action (a tool_use or the assistant's text). A message that only ever thinks is
                    # flushed as "thinking…" when the next message begins (see _flush_turn).
                    content = message.get("content")
                    blocks = content if isinstance(content, list) else []
                    if pending and pending["id"] != message_id:
                        _flush_turn()
                    if should_count:
                        pending = {"id": message_id, "turn": stats["stream_turns"], "blocks": []}
                    if pending:
                        pending["blocks"].extend(b for b in blocks if isinstance(b, dict))
                        # Flush as soon as the turn's *action* is known — a tool_use is the last block
                        # of its message, so the line is final and can print at once. A turn that only
                        # talked or only thought has no such moment; it flushes at the message boundary
                        # above (or at `result`), which costs a little latency but never a wrong line.
                        if any(b.get("type") == "tool_use" for b in pending["blocks"]):
                            _flush_turn()
                elif etype == "result":
                    _flush_turn()
                    stats["cum_turns"] += evt.get("num_turns", 0)
                    cost = evt.get("total_cost_usd", 0) or 0
                    if cost > stats["cost_usd"]:
                        stats["cost_usd"] = cost
                    if evt.get("is_error"):
                        stats["error"] = _result_error(evt)
                    progress(f"Segment done ({evt.get('subtype', '')}): "
                             f"+{evt.get('num_turns', 0)} turns, "
                             f"{stats['cum_turns']}/{cli_max_turns} CLI cap, "
                             f"cost=${stats['cost_usd']:.4f}")
                    if stats["cum_turns"] >= cli_max_turns or hard_stop_at:
                        progress("Turn budget reached — stopping")
                        kill()
                        break
                if hard_stop_at and time.monotonic() - hard_stop_at > _HARD_STOP_GRACE_S:
                    progress("Hard stop grace period elapsed — force killing")
                    kill()
                    break
                if time.monotonic() - start > wall_timeout:
                    stats["timed_out"] = True
                    progress(f"Wall timeout ({wall_timeout}s) exceeded — stopping")
                    kill()
                    break
        finally:
            _flush_turn()   # a stream cut short (kill / timeout) still owes its last turn a line
            proc.wait()
            stats["returncode"] = proc.returncode
            try:   # drain a result event emitted just as the kill fired
                for raw in proc.stdout:
                    evt = record(raw)
                    if evt and evt.get("type") == "result":
                        stats["cum_turns"] += evt.get("num_turns", 0)
                        cost = evt.get("total_cost_usd", 0) or 0
                        if cost > stats["cost_usd"]:
                            stats["cost_usd"] = cost
                        if evt.get("is_error"):
                            stats["error"] = _result_error(evt)
            except (OSError, ValueError):
                pass
            progress(f"CLI exited (code {proc.returncode}), cumulative turns: {stats['cum_turns']}")

    # ---- mode dispatch: the TASK decides where its evaluator runs, so it decides where the agent runs -
    @staticmethod
    def _wants_container(task) -> bool:
        """Whether the agent's CLI session belongs inside the task's container.

        Every docker-mode task does — **including** one whose evaluator is itself a Docker client
        (``evaluation.docker_access``, e.g. ALE-Bench), which is a deliberate exception to this
        scaffold's sandbox story. Such a container carries the host's docker socket, and a socket is
        root on the host, so for that family the task container is an *environment*, not a sandbox: the
        agent can reach the daemon. It is the price of scoring ALE-Bench at all — ``ale_bench`` compiles
        and judges each candidate inside sibling containers it launches on the daemon, and exposes no
        judge service to call instead, so an agent that cannot reach the daemon cannot see its own score.

        Keeping the agent out does not buy safety back. The alternative is the host loop, where the agent
        runs as the operator — who must already have rw on that same socket for ``docker_access`` to
        preflight at all (see :mod:`galapagos.environments.dood`), i.e. the same host-root reach, minus
        the container. It is also strictly worse: on the host the agent's ``run_eval.sh`` and the
        framework's recorded score come from *different* environments, and the two silently disagree.
        """
        return task.eval_mode == "docker"

    def setup(self, task) -> None:
        """Place the agent where the evaluation happens.

        docker mode (the default) → **one** container from the task's own image holds the
        CLI session, and the framework's trusted scorer ``exec``s into that same container to record the
        score (the agent runs non-root; the scorer's ``/eval`` is root-owned and unwritable).
        ``mode: local`` → the session runs as a host subprocess against a host ``SubprocessEvaluator``.
        The task, not the operator, decides.
        """
        self._container_mode = self._wants_container(task)
        if not self._container_mode:
            super().setup(task)
            self._last_agent_snapshot = self.ctx.best
            return
        self._bring_up(task)
        try:
            super().setup(task)
            self._last_agent_snapshot = self.ctx.best
        except Exception:
            self._agent.stop()
            raise

    def _agent_run_summary(self) -> dict:
        """Read the completed CLI receipt for the terminal agent event."""
        candidates: list[Path] = []
        if self.ctx.run_dir:
            candidates.append(Path(self.ctx.run_dir) / "run_summary.json")
        out_dir = getattr(self, "_out_dir", None)
        if out_dir is not None:
            candidates.append(Path(out_dir) / "run_summary.json")
        summary: dict = {}
        for path in candidates:
            try:
                payload = json.loads(path.read_text(encoding="utf-8"))
            except (OSError, json.JSONDecodeError):
                continue
            if isinstance(payload, dict):
                summary = payload
                break
        if not summary.get("native_session_files"):
            session_root = (
                getattr(self, "_claude_config_dir", None)
                or getattr(self, "_codex_home", None)
            )
            if session_root is not None:
                session_root = Path(session_root)
                artifact_root = Path(out_dir) if out_dir is not None else session_root.parent
                files: list[str] = []
                for path in session_root.rglob("*.jsonl"):
                    if not path.is_file():
                        continue
                    try:
                        files.append(str(path.relative_to(artifact_root)))
                    except ValueError:
                        files.append(str(path))
                if files:
                    summary["native_session_files"] = sorted(files)
        summary.setdefault(
            "agent_session_format",
            "codex-native" if self.name == "codex" else "claude-native",
        )
        return summary

    def _loop(self, task) -> None:
        if getattr(self, "_container_mode", None) is None:
            self._container_mode = self._wants_container(task)
        started = time.monotonic()
        session_event = self._emit_event(
            event_type="agent",
            action="start",
            status="completed",
            component_role="proposer",
            component_name=type(self).__name__,
            evolution_track="solution",
            details={
                "agent": "claude_code",
                "model": self.config.proposer.claude_model,
                "execution_mode": "container" if self._container_mode else "host",
                "turn_budget": int(self.general.max_iterations),
            },
        )
        self._active_agent_event_id = session_event
        try:
            if self._container_mode:
                self._loop_container(task)
            else:
                self._loop_host(task)
        except KeyboardInterrupt as exc:
            self._emit_event(
                event_type="agent",
                action="interrupt",
                status="interrupted",
                component_role="proposer",
                component_name=type(self).__name__,
                evolution_track="solution",
                caused_by_event_ids=[session_event] if session_event else [],
                duration_seconds=time.monotonic() - started,
                error=exc,
                details={"agent": "claude_code", **self._agent_run_summary()},
            )
            raise
        except Exception as exc:
            self._emit_event(
                event_type="agent",
                action="fail",
                status="failed",
                component_role="proposer",
                component_name=type(self).__name__,
                evolution_track="solution",
                caused_by_event_ids=[session_event] if session_event else [],
                duration_seconds=time.monotonic() - started,
                error=exc,
                details={"agent": "claude_code", **self._agent_run_summary()},
            )
            raise
        else:
            self._emit_event(
                event_type="agent",
                action="complete",
                status="completed",
                component_role="proposer",
                component_name=type(self).__name__,
                evolution_track="solution",
                caused_by_event_ids=[session_event] if session_event else [],
                duration_seconds=time.monotonic() - started,
                details={
                    "agent": "claude_code",
                    "model": self.config.proposer.claude_model,
                    "turns": int(self.ctx.iteration),
                    **self._agent_run_summary(),
                },
            )
        finally:
            self._active_agent_event_id = None

    # ---- config validation (must run BEFORE anything is started, so a bad flag leaks nothing) -----
    def _validate_config(self) -> tuple[int, int, str | None, str | None]:
        """Validate every proposer knob and return ``(prompt_turn_budget, cli_max_turns, model, effort)``.
        Idempotent, and called from both :meth:`setup` and :meth:`_loop` — the former so a typo never
        leaves a container running, the latter so a direct ``_loop()`` call still validates.
        """
        cfg = self.config.proposer
        prompt_turn_budget = int(self.general.max_iterations)
        if prompt_turn_budget <= 0:
            raise ValueError("claude_code requires general.max_iterations to be positive")
        cli_max_turns = (int(cfg.claude_cli_max_turns) if cfg.claude_cli_max_turns is not None
                         else prompt_turn_budget * 2)
        if cli_max_turns < prompt_turn_budget:
            raise ValueError("proposer.claude_cli_max_turns must be >= general.max_iterations "
                             f"(got {cli_max_turns} < {prompt_turn_budget})")
        model = cfg.claude_model
        if model and not any(model.startswith(p) for p in _CLAUDE_MODEL_PREFIXES):
            raise ValueError(
                f"claude_code only drives Claude models, got {model!r}; set "
                "proposer.claude_model to a claude-*/sonnet/opus/haiku name (or null for the CLI's default)")
        effort = cfg.reasoning_effort
        if effort is not None and effort not in _EFFORT_LEVELS:
            raise ValueError(
                f"claude_code: invalid proposer.reasoning_effort {effort!r} for `claude --effort`; "
                f"use one of {', '.join(_EFFORT_LEVELS)} (or null for the CLI's default)")
        if cfg.claude_wall_timeout_seconds is not None and int(cfg.claude_wall_timeout_seconds) <= 0:
            raise ValueError("proposer.claude_wall_timeout_seconds must be positive when set")
        return prompt_turn_budget, cli_max_turns, model, effort

    def run(self, task=None, **kwargs):
        """Backstop teardown for the agent container.

        ``setup()`` starts it, but the only other teardown site is ``_loop()``'s ``finally`` — so any
        exception raised between them (``write_effective_config``, a resume failure) would strand a
        ``sleep infinity`` container that ``--rm`` never reaps. ``TaskContainer.stop()`` is idempotent.
        """
        try:
            return super().run(task=task, **kwargs)
        finally:
            agent = getattr(self, "_agent", None)
            if agent is not None:
                agent.stop()

    # ---- environment bring-up (Harbor's _prepare: start env → install agent → then run) ----------
    def _bring_up(self, task) -> None:
        """Resolve the task image, layer the CLI onto it, and start ONE container that holds both the
        agent and the trusted scorer.

        Single-container model: the CLI session runs in the task's own image (Harbor's installed-agent
        model) and the framework's trusted scorer ``exec``s into that *same* container to record the
        score-of-record (an *attached* :class:`ContainerEvaluator`, so ``close()`` can never tear the
        agent's live session down). Isolation rests on ownership, not a second container — the agent runs
        **non-root** while ``/eval`` (scorer + data + runner) is **root-owned and unwritable**.

        Paths are deliberately generic: ``HOME``/cwd = ``/workspace`` and the CLI writes its native
        session trace to the DEFAULT ``$HOME/.claude`` (no galapagos-specific ``CLAUDE_CONFIG_DIR``), so
        the recorded trajectory reads like any reference-agent run and stays reusable as training data.

        Called from :meth:`setup`, and lazily from :meth:`_loop` on the ``--resume`` path — which
        ``base_scaffold.run()`` routes through ``_resume()`` instead of ``setup()``, so without this the
        loop would dereference attributes that were never assigned.
        """
        cfg = self.config.proposer
        self._validate_config()

        # resolve(): the workspace under this directory becomes a bind-mount *source*, and Docker reads a
        # relative source as a named volume ("includes invalid characters for a local volume name"), so a
        # relative --output-dir would fail the run at container start.
        run_dir = Path(self.ctx.run_dir).resolve() if self.ctx.run_dir else None
        if run_dir:
            self._out_dir, self._cleanup_out_dir = run_dir, False
        else:
            self._out_dir = Path(tempfile.mkdtemp(prefix="galapagos-claude-docker-"))
            self._cleanup_out_dir = True
        workspace = self._out_dir / "claude_workspace"
        workspace.mkdir(parents=True, exist_ok=True)

        # HOME=/workspace with no CLAUDE_CONFIG_DIR override → the CLI writes its native session JSONL to
        # the default `$HOME/.claude/projects/**.jsonl`, i.e. the host `claude_workspace/.claude`.
        self._claude_config_dir = workspace / ".claude"
        self._claude_config_dir.mkdir(parents=True, exist_ok=True)
        self._auth_source, auth_env = _setup_subscription_auth_docker(self._claude_config_dir)

        spec = task.image_spec()
        # `evaluation.docker_access` (ALE-Bench): the evaluator that will run in this container is itself
        # a Docker client, so the container needs the host socket, the socket's gid, and a path-parity
        # directory (see environments.dood). Resolved BEFORE the image build, not after: a host that
        # cannot grant a daemon must say so in seconds rather than build an image and then score every
        # candidate zero.
        dood_mounts, dood_env, group_add = dood.runtime() if spec.docker_access else ([], {}, [])
        task_hf_runtime = getattr(task, "hf_cache_runtime", None)
        hf_mounts, hf_env = (
            task_hf_runtime() if callable(task_hf_runtime) else _default_hf_cache_runtime()
        )
        task_cache_runtime = getattr(task, "evaluator_cache_runtime", None)
        cache_mounts, cache_env = (
            task_cache_runtime() if callable(task_cache_runtime) else ([], {})
        )

        # The image is the TASK's, not ours. `claude_docker_image` overrides it wholesale for the rare
        # case where an operator has a prebuilt environment they'd rather use.
        task_image = cfg.claude_docker_image or resolve_image(spec)
        agent_image = build_agent_image(task_image, cli_version=cfg.claude_docker_cli_version)
        self._task_image, self._agent_image = task_image, agent_image

        # `dood_env` points HOME *and* TMPDIR at the parity dir, because those are the two roots every
        # daemon-facing path in ale_bench derives from, and the daemon resolves a sibling's volume
        # sources against the HOST. The agent, though, needs HOME=/workspace (the CLI writes its native
        # trace to the default `$HOME/.claude`), so HOME is put back — and ale_bench's cache, the root
        # that would otherwise follow HOME, is pinned explicitly at the path a HOME=<parity> process
        # would have chosen. Both the agent's `run_eval.sh` and the framework's `docker exec` scorer then
        # compute daemon-facing paths that are spelled identically inside and out. Get this wrong and
        # every judge container comes up with empty bind mounts: no source to compile, no case to run,
        # and a zero score with nothing raised.
        env = {
            **dict(spec.env), **dood_env, **auth_env,
            # Keep an allowlisted dataset cache offline even if a card reuses an HF variable.
            **hf_env,
            **cache_env,
            "HOME": WORKSPACE_DIR,
        }
        if spec.docker_access:
            env["ALE_BENCH_CACHE"] = str(dood.parity_dir() / ".cache" / "ale-bench")
        self._agent = TaskContainer.start(
            agent_image,
            name=f"galapagos-claude-{uuid.uuid4().hex[:12]}",
            # The host uid keeps the bind-mounted workspace writable AND keeps the CLI off root (which it
            # refuses to run as with --dangerously-skip-permissions). It is also the trust boundary: the
            # non-root agent cannot rewrite the root-owned /eval scorer the framework records with.
            user=f"{os.getuid()}:{os.getgid()}",
            env=env,
            mounts=[
                (str(workspace), WORKSPACE_DIR, "rw"),
                *hf_mounts,
                *cache_mounts,
                *dood_mounts,
            ],
            group_add=group_add,          # the socket is srw-rw---- root:docker; a non-root uid needs its gid
            gpus=spec.gpus,
            network=cfg.claude_docker_network or spec.network,
            memory=cfg.claude_docker_memory or spec.memory,
            cpus=cfg.claude_docker_cpus or spec.cpus,
            workdir=WORKSPACE_DIR)

        try:
            # The trusted scorer attaches to the agent's own container (a *borrowed* handle, so the
            # scaffold's `finally: evaluator.close()` can never `docker rm -f` the agent's live session).
            # An evaluator carried over from _resume() is a host SubprocessEvaluator; replace it.
            if not isinstance(self.evaluator, ContainerEvaluator):
                self.evaluator = task.container_evaluator(attach_to=self._agent)
            # Provision /eval eagerly (root-owned, group/other-unwritable) so the agent's own run_eval.sh
            # works from turn 1 against the very runner the framework records with.
            provision_task_dir(self._agent, str(task.root), _RUNNER)
        except Exception:
            self._agent.stop()          # never leak the container when bring-up fails
            raise


    # ---- the loop (one CLI session inside the task container) ------------------------------------
    def _loop_container(self, task) -> None:
        cfg = self.config.proposer
        prompt_turn_budget, cli_max_turns, model, effort = self._validate_config()

        # `--resume` routes through base_scaffold._resume(), which never calls setup(). Bring the
        # environment up here so the loop has a container (and a container-mode evaluator) either way.
        if getattr(self, "_agent", None) is None:
            self._bring_up(task)
            self._apply_language_and_suffix(self.ctx.best.content if self.ctx.best else "")

        seed = self.ctx.best                 # set by setup()/resume: the evaluated seed program
        eval_timeout = int(getattr(self.evaluator, "timeout", 120))
        suffix = getattr(self.evaluator, "suffix", ".py")

        out_dir, container = self._out_dir, self._agent
        workspace = out_dir / "claude_workspace"
        log_path, progress_path = out_dir / "claude.log", out_dir / "progress.log"

        # The workspace is a bind mount, so staging happens on the HOST and appears in the container.
        solution_path = workspace / f"solution{suffix}"
        solution_path.write_text(seed.content if seed else "")
        self._write_eval_script_container(workspace, eval_timeout)
        prompt_content = self._write_task_prompt_container(workspace, suffix, prompt_turn_budget, eval_timeout)
        (workspace / ".prompt.txt").write_text(prompt_content)
        self._write_run_script(workspace, cli_max_turns, model, effort)

        wall_timeout = (int(cfg.claude_wall_timeout_seconds)
                        if cfg.claude_wall_timeout_seconds is not None
                        else max(cli_max_turns * (_DEFAULT_TURN_OVERHEAD_S + eval_timeout),
                                 _MIN_WALL_TIMEOUT_S))

        plock = threading.Lock()

        def progress(line: str) -> None:
            log.info("%s", line)
            with plock, open(progress_path, "a") as f:
                f.write(f"[{time.strftime('%H:%M:%S')}] {line}\n")

        progress(f"Claude Code (Docker) run started — task_image={self._task_image}, "
                 f"agent_image={self._agent_image}, "
                 f"model={model or 'CLI default'}, effort={effort or 'CLI default'}, "
                 f"prompt_turn_budget={prompt_turn_budget}, cli_max_turns={cli_max_turns}, "
                 f"billing=subscription ({self._auth_source}), "
                 f"native traces → {self._claude_config_dir / 'projects'}, wall_timeout={wall_timeout}s")

        stats = {"stream_turns": 0, "cum_turns": 0, "cost_usd": 0.0, "error": None,
                 "session_id": None, "timed_out": False, "returncode": None}
        last_content = seed.content if seed else ""
        ckpt_count = 0
        run_start = time.monotonic()
        proc = None
        try:
            with open(log_path, "w") as log_fh:
                proc = container.popen(["bash", f"{WORKSPACE_DIR}/.run.sh"],
                                       stdout=subprocess.PIPE, stderr=log_fh)
                reader = threading.Thread(
                    target=self._pump_events,
                    args=(proc, log_fh, stats, cli_max_turns, wall_timeout, progress),
                    kwargs={"kill_fn": self._kill_agent_session},
                    daemon=True)
                reader.start()
                # Poll the solution file while the session runs; score every new version so the run has
                # real checkpoints even if the session dies mid-way.
                while reader.is_alive():
                    reader.join(timeout=_POLL_INTERVAL_S)
                    try:
                        cur = solution_path.read_text()
                    except OSError:
                        continue
                    if cur == last_content or not cur.strip():
                        continue
                    last_content = cur
                    ckpt_count += 1
                    observed_turns = stats["cum_turns"] or stats["stream_turns"]
                    self._admit(cur, seed, iteration=max(observed_turns, ckpt_count),
                                kind="checkpoint", turn=observed_turns)
                    self._maybe_checkpoint()

            actual_turns = stats["cum_turns"] or stats["stream_turns"]
            _assert_session_did_something(stats, ckpt_count, proc.returncode, log_path)

            try:
                final = solution_path.read_text()
            except OSError:
                final = ""
            if final.strip() and final != last_content:   # written between the last poll and exit
                ckpt_count += 1
                self._admit(final, seed, iteration=max(actual_turns, ckpt_count),
                            kind="final", turn=actual_turns)
            self.ctx.iteration = max(self.ctx.iteration, actual_turns, 1)
            if stats["cost_usd"]:
                self.ctx.record_cost(stats["cost_usd"])

            best = self.ctx.best
            summary = {
                "agent_session_format": "claude-native",
                "model": model, "effort": effort, "billing": "subscription",
                "auth_source": self._auth_source,
                "task_image": self._task_image, "agent_image": self._agent_image,
                "session_id": stats["session_id"],
                "native_session_files": _native_session_files(self._claude_config_dir),
                "prompt_turn_budget": prompt_turn_budget, "cli_max_turns": cli_max_turns,
                "actual_turns": actual_turns, "checkpoints_scored": ckpt_count,
                "timed_out": stats["timed_out"], "return_code": stats["returncode"],
                "cost_usd": round(stats["cost_usd"], 4),
                "wall_seconds": round(time.monotonic() - run_start, 1),
                "baseline_score": seed.scores.get("combined_score") if seed else None,
                "final_score": best.scores.get("combined_score") if best else None,
            }
            (out_dir / "run_summary.json").write_text(json.dumps(summary, indent=2, default=str) + "\n")
            progress(f"Run complete: turns={actual_turns}/{prompt_turn_budget} prompt budget "
                     f"({cli_max_turns} CLI cap), cost=${stats['cost_usd']:.4f}, "
                     f"checkpoints={ckpt_count}, score={summary['final_score']}")
        finally:
            if proc is not None and proc.poll() is None:
                self._kill_agent_session()
                proc.wait()
            # Never ship the operator's subscription credentials inside the trajectory: the login-fallback
            # copy lives in the mounted, preserved `.claude` dir, so drop it now (the OAuth-token path
            # writes no file). The native session JSONL beside it is what stays.
            creds = self._claude_config_dir / ".credentials.json"
            if creds.exists():
                creds.unlink()
            native_sessions = _native_session_files(self._claude_config_dir)
            if native_sessions:
                progress(f"Native Claude Code trace(s) preserved: {', '.join(native_sessions)}")
            else:
                progress("Warning: no native Claude Code session file found under "
                         f"{self._claude_config_dir / 'projects'}")
            # The agent container dies here. The attached evaluator points at it, so the final scoring
            # above must already be done — it is: _admit ran inside the try block.
            container.stop()
            if self._cleanup_out_dir:
                shutil.rmtree(out_dir, ignore_errors=True)

    # ---- helpers ---------------------------------------------------------------------------------
    def _kill_agent_session(self) -> None:
        """Stop the CLI session **without destroying the container**.

        The single container also holds the trusted evaluator the framework still needs for its final
        scoring pass, so killing the container is never the answer here — the run tears it down in
        :meth:`_loop_container`'s ``finally``, after that final ``_admit``. Killing the local
        ``docker exec`` client would leave the in-container process alive, so escalate *inside* the
        container: signal the PID the session recorded, then ``pkill`` the CLI by name (``procps`` is
        installed into the agent image for exactly this).
        """
        for sig in ("TERM", "KILL"):
            if self._agent.kill_process(_AGENT_PID_FILE, sig=sig):
                return
            if self._agent.exec(["pkill", f"-{sig}", "-f", "claude"]).returncode == 0:
                return
        log.warning("could not signal the agent session via %s or pkill; NOT killing the container — it "
                    "still holds the evaluator the run needs to score its final solution", _AGENT_PID_FILE)

    def _write_eval_script_container(self, workspace: Path, timeout: int) -> None:
        """``run_eval.sh`` — the command Claude Code calls to score a candidate.

        It invokes the framework's **own trusted runner** against the task's own ``evaluator.py``, with
        the same cascade configuration the framework uses. That is the guarantee this scaffold exists to
        provide: the number the agent sees is the number that gets recorded. The runner emits its verdict
        after a ``<<<OG_RESULT>>>`` sentinel; everything before it is the evaluator's own chatter.
        """
        if not isinstance(self.evaluator, ContainerEvaluator):
            raise RuntimeError(
                "claude_code scores inside the task's container, so it needs a "
                f"ContainerEvaluator; got {type(self.evaluator).__name__}. Leave scaffold.evaluator "
                "unset — setup() builds the right one.")
        argv = self.evaluator.runner_argv('"$PROGRAM"')
        cmd = " ".join(a if a == '"$PROGRAM"' else shlex.quote(a) for a in argv)
        script = (
            "#!/bin/bash\n"
            "set -uo pipefail\n"
            'PROGRAM="$(realpath "$1")"\n'
            f"cd {EVAL_DIR}\n"
            f'out="$(timeout {timeout} {cmd} 2>&1)"\n'
            'if [[ "$out" == *"<<<OG_RESULT>>>"* ]]; then\n'
            '  printf "%s\\n" "${out#*<<<OG_RESULT>>>}"\n'
            "else\n"
            '  printf "%s\\n" "$out" >&2\n'
            "  exit 1\n"
            "fi\n"
        )
        path = workspace / "run_eval.sh"
        path.write_text(script)
        path.chmod(0o755)

    def _write_task_prompt_container(self, workspace: Path, suffix: str, max_turns: int,
                           eval_timeout: int) -> str:
        """``TASK.md`` — the one-shot brief piped to the CLI, with container paths."""
        content = (
            "You are an AI assistant iteratively improving a program to maximize "
            f"its evaluation score. You have **{max_turns} turns** total. You are running inside "
            "a sandboxed container with no access to the host machine.\n\n"
            "## Current solution\n\n"
            f"`{WORKSPACE_DIR}/solution{suffix}` -- read it, understand it, modify it freely.\n\n"
            "## How to evaluate\n\n"
            "```bash\n"
            f"bash {WORKSPACE_DIR}/run_eval.sh {WORKSPACE_DIR}/solution{suffix}\n"
            "```\n\n"
            "Output is JSON. The `combined_score` field is what you want to maximize "
            f"(higher is better). The evaluator has a **{eval_timeout}s timeout**.\n\n"
            "## Task description\n\n"
            f"{self.ctx.task_context}\n\n"
            "## Instructions\n\n"
            "- Run the evaluator once to confirm the baseline score, then start improving.\n"
            "- After each change, evaluate and decide whether to keep or revert.\n"
            f"- Always keep `{WORKSPACE_DIR}/solution{suffix}` set to your best solution.\n"
            "- Aim to try several distinct approaches within your turn budget.\n"
            f"- Create scratch files only inside `{WORKSPACE_DIR}`.\n"
            f"- `{EVAL_DIR}` holds the task's scorer and its data. "
            "Do not modify it — your score is recomputed independently.\n"
        )
        (workspace / "TASK.md").write_text(content)
        return content

    @staticmethod
    def _write_run_script(workspace: Path, cli_max_turns: int, model: str | None,
                          effort: str | None) -> None:
        """``.run.sh`` — the container-side entrypoint of the session.

        It records its own PID before ``exec``-ing the CLI (``exec`` keeps the PID), which is what lets a
        wall-clock timeout stop the agent without destroying the container.
        """
        claude_cmd = ["claude", "-p", "-", "--max-turns", str(cli_max_turns),
                      "--dangerously-skip-permissions", "--disallowedTools", _DISALLOWED_TOOLS,
                      "--output-format", "stream-json", "--verbose"]
        if model:
            claude_cmd += ["--model", model]
        if effort:
            claude_cmd += ["--effort", effort]
        lines = [
            "#!/bin/bash",
            "set -euo pipefail",
            f"echo $$ > {_AGENT_PID_FILE}",
            "exec " + " ".join(shlex.quote(c) for c in claude_cmd)
            + f" < {WORKSPACE_DIR}/.prompt.txt",
        ]
        path = workspace / ".run.sh"
        path.write_text("\n".join(lines) + "\n")
        path.chmod(0o755)