diff --git a/docs/reference/http-api.rst b/docs/reference/http-api.rst index 8d9c3462..37fdd9c1 100644 --- a/docs/reference/http-api.rst +++ b/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 diff --git a/docs/user-guide/recipes/reefine.rst b/docs/user-guide/recipes/reefine.rst index a9da9ac0..1aa3fe99 100644 --- a/docs/user-guide/recipes/reefine.rst +++ b/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 \ diff --git a/pyproject.toml b/pyproject.toml index 15484736..e3aeff18 100644 --- a/pyproject.toml +++ b/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 = [ diff --git a/reef/harness/episodes/e2b.py b/reef/harness/episodes/e2b.py index 1d5bc753..77f66d66 100644 --- a/reef/harness/episodes/e2b.py +++ b/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) diff --git a/reef/harness/episodes/e2b_relay.py b/reef/harness/episodes/e2b_relay.py index 8d55ff8f..8b794afc 100644 --- a/reef/harness/episodes/e2b_relay.py +++ b/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 diff --git a/reef/recipe/reefine/agent.py b/reef/recipe/reefine/agent.py index 84a226db..9de73834 100644 --- a/reef/recipe/reefine/agent.py +++ b/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 diff --git a/reef/service/page_chrome.py b/reef/service/page_chrome.py index e9abb9b1..ffe83b06 100644 --- a/reef/service/page_chrome.py +++ b/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", diff --git a/reef/service/release_page.py b/reef/service/release_page.py index f8f01963..5e6dab81 100644 --- a/reef/service/release_page.py +++ b/reef/service/release_page.py @@ -69,6 +69,7 @@ "selected": "Passed the checks and was published as the served head.", "rejected": "Did not pass the checks. The head stayed where it was.", "skipped": "No candidate reached the evaluation, so nothing changed.", + "failed": "The step failed before evaluation. Nothing was published; see the error below before retrying.", "creation": "The tree this scenario started from, before any step ran.", "promote": "A person promoted a pending release, which now serves.", "rollback": "A person moved the head back to an earlier release.", @@ -157,7 +158,7 @@ def mutations_of(metrics: Mapping[str, Any] | None) -> list[Mapping[str, Any]]: def result_of(row: Mapping[str, Any], rows: Sequence[Mapping[str, Any]] = ()) -> str: - """The row's result: pending, selected, rejected, skipped, else the operation (creation, promote, rollback). + """The row's result, including failed proposals that published no release. A pending row stays pending in the catalog after a person promotes it; the promote is a later row naming it in ``rollback_target_release_id``, @@ -173,6 +174,11 @@ def result_of(row: Mapping[str, Any], rows: Sequence[Mapping[str, Any]] = ()) -> if isinstance(metrics.get("selected"), bool): return "selected" if metrics["selected"] else "rejected" if metrics.get("skipped"): + notes = metrics.get("proposal_notes") + failure = notes.get("failure") if isinstance(notes, Mapping) else None + error = metrics.get("error") + if any(isinstance(value, str) and value.strip() for value in (failure, error)): + return "failed" return "skipped" return str(row.get("operation") or "unknown") @@ -184,7 +190,7 @@ def served_step(rows: Sequence[Mapping[str, Any]]) -> int | None: pending win or a failed evaluation makes the wrong one: those rows publish nothing, and a rejected or skipped row carries the head's id.""" for index in range(len(rows) - 1, -1, -1): - if result_of(rows[index]) not in ("pending", "rejected", "skipped"): + if result_of(rows[index]) not in ("pending", "rejected", "skipped", "failed"): return index return None @@ -198,7 +204,7 @@ def before_release_id(row: Mapping[str, Any]) -> str | None: if selection_result in ("selected", "pending"): parent = row.get("parent_release_id") return str(parent) if parent else None - if selection_result in ("rejected", "skipped"): + if selection_result in ("rejected", "skipped", "failed"): return str(row.get("release_id") or "") or None return None @@ -501,7 +507,7 @@ def _setup(row: Mapping[str, Any], metrics: Mapping[str, Any], rows: Sequence[Ma requires = request.get("requires") own = [item for item in requires if isinstance(item, Mapping)] if isinstance(requires, Sequence) else [] carried: list[Mapping[str, Any]] = [] - if result_of(row) not in ("rejected", "skipped"): + if result_of(row) not in ("rejected", "skipped", "failed"): names = {item.get("name") for item in own} carried = [item for item in required_by(rows, row.get("release_id")) if item.get("name") not in names] refused = [ @@ -546,7 +552,7 @@ def _chain( ) -> str: release_id = row.get("release_id") selection_result = result_of(row) - if selection_result in ("rejected", "skipped"): + if selection_result in ("rejected", "skipped", "failed"): # The row carries the head's id and published nothing, so the head's parent and children are not its own. if selection_result == "rejected": ran_on = "the head at this step; the candidate published nothing" diff --git a/reef/service/request_page.py b/reef/service/request_page.py index 4cff1f73..be6b6cf9 100644 --- a/reef/service/request_page.py +++ b/reef/service/request_page.py @@ -240,6 +240,8 @@ def meaning(selection_result: str, row: Mapping[str, object], metrics: Mapping[s return f"did not pass the checks ({reason or 'the checks failed'}); nothing changed: rephrase or split the request" if selection_result == "skipped": return f"produced no change ({metrics.get('skipped')}); nothing changed" + if selection_result == "failed": + return "The step failed before evaluation. Nothing was published; see the error below before retrying." return f"the step ended as {selection_result}" @@ -375,7 +377,7 @@ def build_request_page( else: metrics = rows[step].get("metrics") metrics = metrics if isinstance(metrics, Mapping) else {} - change_label = "Proposed changes" if state in ("pending", "rejected", "skipped") else "What changed" + change_label = "Proposed changes" if state in ("pending", "rejected", "skipped", "failed") else "What changed" body = ( f'
\n

Result

\n{result_html(step, rows, link_query)}
\n' f'
\n

{change_label}

\n{what_changed(metrics)}
\n' diff --git a/tests/reef_service/test_e2b_tunnel.py b/tests/reef_service/test_e2b_tunnel.py index 55b996c2..65c4af1d 100644 --- a/tests/reef_service/test_e2b_tunnel.py +++ b/tests/reef_service/test_e2b_tunnel.py @@ -2,6 +2,7 @@ from __future__ import annotations +import http.client import io import json import socket @@ -15,12 +16,14 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from types import SimpleNamespace +from unittest.mock import Mock import pytest from reef.harness.episodes.e2b import ( RELAY_SCRIPT, E2BExecutor, + E2BSession, TunnelPump, pack, remote_command, @@ -28,7 +31,87 @@ template_alias, unpack, ) -from reef.harness.episodes.executor import SandboxUnavailable +from reef.harness.episodes.e2b_relay import Relay +from reef.harness.episodes.executor import EpisodeLaunchError, EpisodeTimeout, ProcessOutcome, SandboxUnavailable + + +@pytest.mark.unit +@pytest.mark.parametrize("disconnects", [1, 3]) +def test_command_disconnect_reconnects_without_starting_the_command_again(tmp_path, monkeypatch, disconnects): + pytest.importorskip("e2b") + rpc = pytest.importorskip("connectrpc.errors") + from connectrpc.code import Code + + sandbox = Mock() + command = sandbox.commands.run.return_value + command.pid = 42 + failure = rpc.ConnectError(Code.INTERNAL, "peer closed connection without sending TLS close_notify") + command.wait.side_effect = failure + resumed = sandbox.commands.connect.return_value + resumed.pid = 42 + if disconnects == 1: + resumed.wait.return_value = SimpleNamespace(exit_code=0, stdout="complete output", stderr="") + else: + resumed.wait.side_effect = failure + session = E2BSession(sandbox) + pushed, pulled = [], [] + monkeypatch.setattr(session, "push", pushed.append) + monkeypatch.setattr(session, "pull", pulled.append) + if disconnects == 1: + outcome = session.launch(["pi"], root=tmp_path, workspace=tmp_path, env={}, timeout=60) + assert outcome == ProcessOutcome(0, "complete output", "") + else: + with pytest.raises(EpisodeLaunchError, match="connection"): + session.launch(["pi"], root=tmp_path, workspace=tmp_path, env={}, timeout=60) + sandbox.commands.kill.assert_called_once_with(42) + sandbox.commands.run.assert_called_once() + assert sandbox.commands.run.call_args.kwargs["background"] is True + assert sandbox.commands.connect.call_count == min(disconnects, 2) + for call in sandbox.commands.connect.call_args_list: + assert call.args == (42,) + assert 0 < call.kwargs["timeout"] < 60 + assert pushed == pulled == [tmp_path] + + +@pytest.mark.unit +def test_command_timeout_kills_the_process_before_copying_its_files(tmp_path, monkeypatch): + e2b = pytest.importorskip("e2b") + sandbox = Mock() + command = sandbox.commands.run.return_value + command.pid = 42 + command.wait.side_effect = e2b.TimeoutException("deadline exceeded") + session = E2BSession(sandbox) + monkeypatch.setattr(session, "push", lambda root: None) + + def pull(root): + sandbox.commands.kill.assert_called_once_with(42) + + monkeypatch.setattr(session, "pull", pull) + with pytest.raises(EpisodeTimeout): + session.launch(["pi"], root=tmp_path, workspace=tmp_path, env={}, timeout=60) + sandbox.commands.connect.assert_not_called() + + +@pytest.mark.unit +def test_reconnecting_does_not_extend_the_command_deadline(tmp_path, monkeypatch): + pytest.importorskip("e2b") + rpc = pytest.importorskip("connectrpc.errors") + from connectrpc.code import Code + + import reef.harness.episodes.e2b as executor + + monkeypatch.setattr(executor, "time", SimpleNamespace(monotonic=Mock(side_effect=[0.0, 61.0]))) + sandbox = Mock() + process = sandbox.commands.run.return_value + process.pid = 42 + process.wait.side_effect = rpc.ConnectError(Code.INTERNAL, "connection interrupted") + session = E2BSession(sandbox) + monkeypatch.setattr(session, "push", lambda root: None) + monkeypatch.setattr(session, "pull", lambda root: None) + with pytest.raises(EpisodeTimeout): + session.launch(["pi"], root=tmp_path, workspace=tmp_path, env={}, timeout=60) + sandbox.commands.connect.assert_not_called() + sandbox.commands.kill.assert_called_once_with(42) def free_port() -> int: @@ -80,6 +163,14 @@ def do_POST(self) -> None: self.send_response(204) self.send_header("Content-Length", "0") self.end_headers() + elif self.path == "/broken": + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Transfer-Encoding", "chunked") + self.end_headers() + self.wfile.write(b"b\r\ndata: one\n\n\r\n") + self.wfile.flush() + self.close_connection = True def log_message(self, *args) -> None: pass @@ -175,6 +266,58 @@ def test_the_tunnel_answers_no_one_without_the_secret(tunnel) -> None: assert refused.value.code == 403 +@pytest.mark.unit +def test_an_interrupted_upstream_stream_is_not_reported_as_complete(tunnel) -> None: + _, local_port, _ = tunnel + with ( + urllib.request.urlopen(post(f"http://127.0.0.1:{local_port}/broken", {}), timeout=10) as response, + pytest.raises(http.client.IncompleteRead), + ): + response.read() + + +@pytest.mark.unit +@pytest.mark.parametrize("lost_acknowledgements", [1, 3]) +def test_a_lost_reply_acknowledgement_does_not_duplicate_stream_content(lost_acknowledgements: int) -> None: + gateway = Gateway() + relay = Relay("test-secret") + handler = relay.tunnel_handler() + lost: list[int] = [] + + class LoseAcknowledgement(handler): + def answer(self, status: int, body: bytes = b"") -> None: + if self.path.startswith("/reply/") and status == 200 and len(lost) < lost_acknowledgements: + lost.append(status) + self.close_connection = True + self.connection.shutdown(socket.SHUT_RDWR) + return + super().answer(status, body) + + local_server = ThreadingHTTPServer(("127.0.0.1", 0), relay.local_handler()) + tunnel_server = ThreadingHTTPServer(("127.0.0.1", 0), LoseAcknowledgement) + for server in (local_server, tunnel_server): + server.daemon_threads = True + threading.Thread(target=server.serve_forever, daemon=True).start() + pump = TunnelPump(f"http://127.0.0.1:{tunnel_server.server_port}", "test-secret", gateway.port, pollers=1) + try: + pump.start(timeout=10) + gateway.release.set() + request = post(f"http://127.0.0.1:{local_server.server_port}/stream", {}) + with urllib.request.urlopen(request, timeout=10) as response: + if lost_acknowledgements == 1: + assert response.read() == b"data: one\n\ndata: two\n\n" + else: + with pytest.raises(http.client.IncompleteRead): + response.read() + assert len(lost) == lost_acknowledgements + finally: + pump.stop() + for server in (local_server, tunnel_server): + server.shutdown() + server.server_close() + gateway.close() + + @pytest.mark.unit def test_a_root_crosses_as_it_is_and_the_command_names_its_sandbox_path(tmp_path: Path) -> None: root = tmp_path / "reef-proposer-abc" diff --git a/tests/reef_service/test_reefine_agent.py b/tests/reef_service/test_reefine_agent.py index b9372c3c..fceadfa3 100644 --- a/tests/reef_service/test_reefine_agent.py +++ b/tests/reef_service/test_reefine_agent.py @@ -242,6 +242,8 @@ def post(url, body): {{"model": "not/real", "input": sys.argv[-1]}}) text = "spoke with status %d" % status event = {{"type": "message", "message": {{"role": "assistant", "content": [{{"type": "text", "text": text}}]}}}} + if mode == "stream-error": + event["message"].update(stopReason="error", errorMessage="Stream ended without finish_reason") (sessions / "s.jsonl").write_text(json.dumps(event) + "\\n") """ ) @@ -380,6 +382,19 @@ def test_an_agent_that_changes_nothing_or_runs_out_of_time_hands_back_no_mutatio assert slow.mutations == () and "past its" in slow.notes["failure"] +@pytest.mark.unit +def test_an_agent_with_a_failed_model_response_does_not_publish_its_partial_changes(tmp_path, upstream) -> None: + record: list[dict] = [] + host = agent_host(tmp_path, record) + Path(host.binary).with_name("mode").write_text("stream-error") + proposal = AgentProposer(provider_of(upstream))( + NODES, (), served_models(upstream, record, host), requests=[{"text": "x"}], entries=ENTRIES, agent_host=host + ) + assert proposal.mutations == () + assert proposal.notes["agent"]["exit_code"] == 0 + assert proposal.notes["failure"] == "the agent's model response failed: Stream ended without finish_reason" + + @pytest.mark.unit def test_without_a_request_or_an_agent_the_text_proposer_answers(monkeypatch) -> None: seen = [] diff --git a/tests/reef_service/test_release_page.py b/tests/reef_service/test_release_page.py index a09e6b1f..be928b85 100644 --- a/tests/reef_service/test_release_page.py +++ b/tests/reef_service/test_release_page.py @@ -830,6 +830,10 @@ def test_the_result_names_why_the_proposer_produced_nothing_when_the_step_record page = build_release_page(1, [creation, row]) page.encode("ascii") # A failure alone adds no Design or Review section; the Result section names it after the skip, escaped. + assert result_of(row) == "failed" + assert served_step([creation, row]) == 0 + assert before_release_id(row) == "rel-0" + assert 'Failed' in page assert _sections(page) == ["Why", "What changed", "Result", "Setup", "Chain"] selection_result = _section(page, "Result") assert "
Skipped
no proposal
" in selection_result @@ -841,6 +845,7 @@ def test_the_result_names_why_the_proposer_produced_nothing_when_the_step_record # Without the note, or with one that is not text, there is no such row. for notes in ({}, {"failure": " "}, {"failure": 3}): without = {**row, "metrics": {**row["metrics"], "proposal_notes": notes}} + assert result_of(without) == "skipped" assert "Proposer failure" not in build_release_page(1, [creation, without]) # A design written before the reply came to nothing keeps its section beside the row. designed = {**row, "metrics": {**row["metrics"], "proposal_notes": {"design": "A tool.", "failure": failure}}} diff --git a/tests/reef_service/test_request_page.py b/tests/reef_service/test_request_page.py index dc35c250..27bb2efe 100644 --- a/tests/reef_service/test_request_page.py +++ b/tests/reef_service/test_request_page.py @@ -30,6 +30,7 @@ settled_step, ) from reef.train.cordis_backend import Mutation, StepProgress +from reef.train.cordis_backend.strategies import StepProposal MODULE = Path(__file__).parents[2] / "reef" / "service" / "request_page.py" REFRESH = f'' @@ -238,11 +239,11 @@ def test_a_skipped_request_shows_why_the_proposer_produced_nothing_and_what_the_ } skipped = _row(_answered(skipped="no proposal", proposal_notes=notes), release_id="rel-0") page = build_request_page(_record(compacted_at=1_050.0), [CREATION, skipped], now=1_100.0) - assert REFRESH not in page and 'No changes' in page + assert REFRESH not in page and 'Failed' in page assert _sections(page) == ["Request", "Result", "Proposed changes", "Review", "Design"] assert "

one rules entry

" in _section(page, "Design") selection_result = _section(page, "Result") - assert "produced no change (no proposal); nothing changed" in selection_result + assert "The step failed before evaluation. Nothing was published" in selection_result assert ( "

Proposer failure

model call failed after 60.0 s (max_tokens=16384): timeout

" in selection_result @@ -260,7 +261,7 @@ def test_a_skipped_request_shows_why_the_proposer_produced_nothing_and_what_the_ assert "

Review

" not in page and _sections(page)[-1] == "Design" and "

How to use

" not in page failed = _row(_answered(skipped="instruction failed", error="RuntimeError: poison proposer"), release_id="rel-0") page = build_request_page(_record(compacted_at=1_050.0), [CREATION, failed], now=1_100.0) - assert "produced no change (instruction failed)" in page + assert 'Failed' in page assert "

Error

RuntimeError: poison proposer

" in page @@ -331,10 +332,12 @@ def test_the_page_module_is_ascii_and_the_builder_escapes_the_request_the_notes_ assert "