fix(reefine): recover E2B disconnects and surface failed proposals

docs/reference/http-api.rst

@@ -802,7 +802,7 @@ id settles the page: the reload stops and Progress gives way to Result
 action, a failed instruction's ``error``, ``proposal_notes.failure`` as
 ``proposer failure``, the release id and a link to the version page), What
 changed (each mutation's op, id and kind; labeled Proposed changes for
-pending, rejected or skipped steps) and, when the step recorded a review,
+pending, rejected, skipped or failed steps) and, when the step recorded a review,
 Review (its result and the points it left uncovered). Published and pending
 results show the session command to install or promote when the person is
 ready, alongside a link to the version page. An unknown
@@ -813,7 +813,8 @@ as JSON (``Cache-Control: no-store``), for a client that polls rather than a
 browser that renders: ``request_id``, ``settled``, ``step`` (the step the
 row landed as once it settles, else null), ``state`` (the page's own
 ``queued``, ``proposing``, ``evaluating``, ``running`` or ``settling``, and
-the settled row's result once a row answers the request), ``meaning`` (the
+the settled row's result once a row answers the request, including ``failed``
+when a skipped step records a proposer failure or execution error), ``meaning`` (the
 words the page prints beside the state, null once settled), and, while a
 step holds this request, ``started_at``, ``episodes_total``,
 ``step_record`` and ``activity`` (the Activity lines oldest first, each

docs/user-guide/recipes/reefine.rst

@@ -183,6 +183,13 @@ Isolation (``evolution.proposer_agent.sandbox``, or ``REEF_PROPOSER_SANDBOX``):
   Node 22) on first use, in about a minute. The sandbox's own user can reach
   root in it; nothing there holds a key.
 
+  If the command connection drops, Reef attempts to reconnect to the same
+  process twice within the original time limit; it does not rerun the command.
+  The tunnel retries interrupted response transfers without duplicating chunks.
+  A response that cannot be completed remains a failed stream. Failed proposal
+  steps appear as ``Failed`` on the request and release pages, with the recorded
+  error, and publish no release. After resolving the error, submit the request again.
+
   .. code:: bash
 
      E2B_API_KEY=e2b_... reef serve --recipe reefine \

pyproject.toml

@@ -44,7 +44,7 @@ tinker = [
 ]
 postgres = ["psycopg[binary]>=3.1,<4"]
 # The agent proposer's E2B cloud sandbox (evolution.proposer_agent.sandbox: e2b).
-e2b = ["e2b>=2.46"]
+e2b = ["e2b>=2.46", "connectrpc>=0.11.1,<0.12"]
 # Provider-neutral experiment tracking. Non-Slime training backends can opt in
 # without installing the Slime/GPU dependency stack.
 wandb = [

reef/harness/episodes/e2b.py

@@ -41,12 +41,13 @@
 import socket
 import tarfile
 import threading
+import time
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass, field, replace
 from pathlib import Path
 from typing import Any
 
-from aiohttp import ClientSession, ClientTimeout
+from aiohttp import ClientConnectionError, ClientPayloadError, ClientResponseError, ClientSession, ClientTimeout
 
 from reef.harness.episodes.executor import (
     EpisodeExecutor,
@@ -300,21 +301,50 @@ def launch(
         writable_paths: Sequence[Path] = (),
         readonly_paths: Sequence[Path] = (),
     ) -> ProcessOutcome:
+        from connectrpc.code import Code
+        from connectrpc.errors import ConnectError
         from e2b import CommandExitException, TimeoutException
+        from e2b.sandbox_sync.commands.command_handle import CommandHandle
 
         command, envs = remote_command(argv, env, root)
         cwd = f"{remote_root(root)}/{workspace.relative_to(root).as_posix()}"
+        process: CommandHandle | None = None
+        finished = False
         try:
             self.push(root)
-            result = self.sandbox.commands.run(command, envs=envs, cwd=cwd, timeout=timeout)
-            return ProcessOutcome(result.exit_code, result.stdout, result.stderr)
+            deadline = time.monotonic() + timeout
+            process = self.sandbox.commands.run(command, envs=envs, cwd=cwd, timeout=timeout, background=True)
+            process_id = process.pid
+            for attempt in range(3):
+                try:
+                    if attempt:
+                        remaining_seconds = deadline - time.monotonic()
+                        if remaining_seconds <= 0:
+                            raise TimeoutException("the command deadline expired while reconnecting")
+                        process = self.sandbox.commands.connect(process_id, timeout=remaining_seconds)
+                    result = process.wait()
+                    finished = True
+                    return ProcessOutcome(result.exit_code, result.stdout, result.stderr)
+                except ConnectError as exc:
+                    if exc.code not in (Code.INTERNAL, Code.UNKNOWN, Code.UNAVAILABLE) or attempt == 2:
+                        raise
+                    # Reattach to the same PID: replaying a command could repeat its side effects.
+                    process.disconnect()
+                    logger.warning("the E2B command connection dropped; reconnecting to process %s", process_id)
+            raise EpisodeLaunchError("the E2B command connection could not be restored")
         except CommandExitException as exc:
+            finished = True
             return ProcessOutcome(exc.exit_code, exc.stdout, exc.stderr)
         except TimeoutException as exc:
             raise EpisodeTimeout(f"the process ran past its {timeout:g} s limit in the E2B sandbox") from exc
         except Exception as exc:
             raise EpisodeLaunchError(f"the E2B sandbox could not run the process: {exc}") from exc
         finally:
+            if process is not None and not finished:
+                try:
+                    self.sandbox.commands.kill(process.pid)
+                except Exception as exc:
+                    logger.warning("could not stop the E2B process %s: %s", process.pid, exc)
             try:
                 self.pull(root)
             except Exception as exc:
@@ -427,10 +457,32 @@ async def poll(self, tunnel: ClientSession, local: ClientSession) -> None:
 
     async def serve(self, tunnel: ClientSession, local: ClientSession, request: Mapping[str, Any]) -> None:
         reply = f"{self.url}/reply/{request['id']}"
+        sequence = 0
 
         async def send(piece: dict[str, Any]) -> None:
-            async with tunnel.post(reply, json=piece, headers=self.headers) as response:
-                await response.read()
+            nonlocal sequence
+            for attempt in range(3):
+                try:
+                    async with tunnel.post(
+                        reply,
+                        json={**piece, "sequence": sequence},
+                        headers=self.headers,
+                        timeout=ClientTimeout(total=30),
+                    ) as response:
+                        # A final piece can reach the client before its acknowledgement is lost.
+                        if response.status == 410 and piece.get("end") and attempt:
+                            return
+                        response.raise_for_status()
+                        await response.read()
+                    sequence += 1
+                    return
+                except ClientResponseError as exc:
+                    if (exc.status < 500 and exc.status != 429) or attempt == 2:
+                        raise
+                except (ClientConnectionError, ClientPayloadError, TimeoutError):
+                    if attempt == 2:
+                        raise
+                await asyncio.sleep(0.25 * (attempt + 1))
 
         headers = {
             name: value for name, value in (request.get("headers") or {}).items() if name.lower() not in HOP_BY_HOP
@@ -456,14 +508,15 @@ async def send(piece: dict[str, Any]) -> None:
             await send({**(head or {}), "data": "", "end": True})
         except asyncio.CancelledError:
             raise
-        except Exception as exc:
+        except (ClientConnectionError, ClientPayloadError, ClientResponseError, TimeoutError) as exc:
+            logger.warning("the E2B tunnel response was interrupted: %s", exc)
             try:
                 if head is None:
                     error = base64.b64encode(f"the Reef host did not answer: {exc}".encode()).decode()
                     await send({"status": 502, "headers": {"Content-Type": "text/plain"}, "data": error, "end": True})
                 else:
-                    await send({"data": "", "end": True})
-            except Exception as failure:
+                    await send({"data": "", "end": True, "error": True})
+            except (ClientConnectionError, ClientPayloadError, ClientResponseError, TimeoutError) as failure:
                 logger.debug("the E2B tunnel lost a reply: %s", failure)
 
 

reef/harness/episodes/e2b_relay.py

@@ -43,6 +43,7 @@ def __init__(self, method: str, path: str, headers: dict[str, str], body: bytes)
             "body": base64.b64encode(body).decode(),
         }
         self.pieces: queue.Queue[dict] = queue.Queue()
+        self.sequence = 0
 
 
 class Relay:
@@ -56,8 +57,8 @@ def local_handler(self) -> type[BaseHTTPRequestHandler]:
         relay = self
 
         class Local(BaseHTTPRequestHandler):
-            # A close-delimited answer, so a stream passes piece by piece without a length or chunking.
-            protocol_version = "HTTP/1.0"
+            # Chunk framing makes an interrupted answer distinguishable from a complete one.
+            protocol_version = "HTTP/1.1"
 
             def relay_request(self) -> None:
                 length = int(self.headers.get("Content-Length") or 0)
@@ -73,21 +74,31 @@ def relay_request(self) -> None:
                     self.send_error(504, "Reef did not answer through the tunnel")
                     return
                 try:
-                    self.send_response(int(first.get("status") or 502))
+                    status = int(first.get("status") or 502)
+                    self.send_response(status)
                     for name, value in (first.get("headers") or {}).items():
                         if name.lower() not in ("content-length", "transfer-encoding", "connection"):
                             self.send_header(name, value)
+                    chunked = status not in (204, 304)
+                    if chunked:
+                        self.send_header("Transfer-Encoding", "chunked")
                     self.end_headers()
                     piece = first
                     while True:
                         data = base64.b64decode(piece.get("data") or "")
-                        if data:
-                            self.wfile.write(data)
+                        if data and chunked:
+                            self.wfile.write(b"%x\r\n%s\r\n" % (len(data), data))
                             self.wfile.flush()
                         if piece.get("end"):
+                            if piece.get("error"):
+                                self.close_connection = True
+                            elif chunked:
+                                self.wfile.write(b"0\r\n\r\n")
+                                self.wfile.flush()
                             return
                         piece = pending.pieces.get(timeout=ANSWER_SECONDS)
                 except (BrokenPipeError, ConnectionResetError, queue.Empty):
+                    self.close_connection = True
                     return
                 finally:
                     with relay.lock:
@@ -147,13 +158,27 @@ def do_POST(self) -> None:
                 if not self.path.startswith("/reply/"):
                     self.answer(404)
                     return
+                piece = json.loads(raw or b"{}")
+                sequence = piece.get("sequence")
+                if not isinstance(sequence, int) or isinstance(sequence, bool) or sequence < 0:
+                    self.answer(400)
+                    return
                 with relay.lock:
                     pending = relay.open.get(self.path[len("/reply/") :])
-                if pending is None:
-                    self.answer(410)
-                    return
-                pending.pieces.put(json.loads(raw or b"{}"))
-                self.answer(200)
+                    if pending is None:
+                        status = 410
+                    elif piece.get("error"):
+                        # An abort must arrive even if the last data piece's acknowledgements were all lost.
+                        pending.pieces.put(piece)
+                        status = 200
+                    elif sequence > pending.sequence:
+                        status = 409
+                    else:
+                        if sequence == pending.sequence:
+                            pending.pieces.put(piece)
+                            pending.sequence += 1
+                        status = 200
+                self.answer(status)
 
             def log_message(self, format: str, *args: object) -> None:
                 pass

reef/recipe/reefine/agent.py

@@ -471,7 +471,7 @@ def answer_with_agent(
             env = {"REEF_PROPOSER_URL": gateway.base_url, **trial_env(gateway.base_url)}
             run.session = open_session(host, gateway.port)
             host.calls.note("proposer", f"the coding agent started on the request (at most {host.timeout_s:g} s)")
-            outcome, _ = launch_pi(
+            outcome, trajectory = launch_pi(
                 host,
                 run.executor(),
                 rendered_files(agent_nodes, binding, host),
@@ -483,11 +483,19 @@ def answer_with_agent(
             agent["exit_code"] = outcome.exit_code
             if outcome.exit_code != 0:
                 agent["stderr_tail"] = outcome.stderr[-MAX_STDERR_CHARS:]
+            # Pi can exit zero after a failed model response; its final assistant event carries the error.
+            for event in reversed(trajectory):
+                message = event.get("message")
+                if not isinstance(message, Mapping) or message.get("role") != "assistant":
+                    continue
+                if message.get("stopReason") in ("error", "aborted"):
+                    agent["error"] = str(message.get("errorMessage") or "the model response was interrupted")
+                break
         except EpisodeTimeout:
             agent["timed_out"] = True
         except EpisodeLaunchError as error:
-            host.calls.note("proposer", f"the coding agent could not start: {error}", failed=True)
-            return StepProposal((), {"failure": f"the agent could not start: {error}"})
+            host.calls.note("proposer", f"the coding agent run failed: {error}", failed=True)
+            return StepProposal((), {"failure": f"the agent run failed: {error}"})
         finally:
             if run.session is not None:
                 run.session.close()
@@ -499,7 +507,7 @@ def answer_with_agent(
         host.calls.note(
             "proposer",
             f"the coding agent {ended} after {agent['seconds']:g} s and {run.trials} trials; reading its workspace",
-            failed=bool(agent.get("timed_out")) or agent.get("exit_code", 0) != 0,
+            failed=bool(agent.get("timed_out") or agent.get("error")) or agent.get("exit_code", 0) != 0,
         )
         return read_answer(workspace, request, models, nodes, entries, agent)
     finally:
@@ -543,6 +551,8 @@ def read_answer(
         notes["design"] = design[: evolution._DESIGN_CHARS]
     if agent.get("timed_out"):
         return StepProposal((), {**notes, "failure": f"the agent ran past its {agent['seconds']:g} s limit"})
+    if agent.get("error"):
+        return StepProposal((), {**notes, "failure": f"the agent's model response failed: {agent['error']}"})
     mutations, problems = workspace_mutations(workspace, entries, nodes)
     if problems:
         notes["unread"] = problems

reef/service/page_chrome.py

@@ -61,13 +61,13 @@
 .status:before{content:"";width:6px;height:6px;border-radius:50%;background:currentColor}
 .tone-selected,.tone-promoted{--status:var(--good);--status-bg:var(--good-bg)}
 .tone-pending,.tone-skipped{--status:var(--warn);--status-bg:var(--warn-bg)}
-.tone-rejected{--status:var(--bad);--status-bg:var(--bad-bg)}
+.tone-rejected,.tone-failed{--status:var(--bad);--status-bg:var(--bad-bg)}
 .tone-queued,.tone-proposing,.tone-evaluating,.tone-running,.tone-settling,.tone-creation,.tone-promote,
 .tone-rollback,.tone-recovery,.tone-unknown{--status:var(--accent);--status-bg:var(--soft)}
 .status{color:var(--status);background:var(--status-bg);border-color:transparent}
 .selected,.promoted,.complete{color:var(--good)}
 .pending,.skipped,.partial,.queued,.proposing,.evaluating,.running,.settling{color:var(--warn)}
-.rejected{color:var(--bad)}.creation,.promote,.rollback,.recovery,.unknown{color:var(--accent)}
+.rejected,.failed{color:var(--bad)}.creation,.promote,.rollback,.recovery,.unknown{color:var(--accent)}
 /* A status word inside a pill takes the pill's tone, which the hero and the summary set. */
 .status span{color:inherit}
 .card{background:var(--card);border:1px solid var(--line);border-radius:12px;padding:28px;min-width:0;box-shadow:var(--shadow)}
@@ -106,6 +106,7 @@
     "pending": "Ready for review",
     "rejected": "Not selected",
     "skipped": "No changes",
+    "failed": "Failed",
     "complete": "Complete",
     "partial": "Partial",
     "creation": "Starting point",