diff --git a/vero/src/vero/harbor/_stream_patch/sitecustomize.py b/vero/src/vero/harbor/_stream_patch/sitecustomize.py new file mode 100644 index 00000000..a61f933f --- /dev/null +++ b/vero/src/vero/harbor/_stream_patch/sitecustomize.py @@ -0,0 +1,366 @@ +"""Make Modal's stdio reconnect budget per-outage inside the harbor subprocess. + +Harbor reads a whole agent phase through one Modal stdio stream. A measured +optimizer phase held that single stream open for 3h19m. + +Modal gives the stream a reconnect budget of ``stream_stdio_max_retries`` and +then never replenishes it. In modal 1.5.3, +``modal/_utils/task_command_router_client.py:769`` sets +``num_retries_remaining = self.stream_stdio_max_retries`` once, above the +``while True``; the success path at :801-802 resets only the backoff delay +(``# Reset retry backoff after any successful chunk.``), not the count. Every +reconnect over the life of the stream draws down the same counter, so what reads +like "retries per problem" is really a LIFETIME cap on a generator that lives as +long as the exec. Hours of healthy streaming in between earn nothing back. + +That is the defect. An earlier version of this patch raised the count instead, +which treats a symptom: any finite lifetime cap is the wrong shape for a stream +measured in hours, because it is spent by unrelated blips spread across the run +rather than by the outage that actually kills it. The fix is per-outage +semantics, and the correct edit is one line next to :802 restoring the count +alongside the delay. + +Two seams, applied together: + +1. The three numeric knobs are constructor keywords, and + ``TaskCommandRouterClient._connect`` never passes them, so nothing in Modal's + public surface can change them: not ``init()``, not ``init_v2()``, and + ``modal/config.py`` has no entry so no environment variable either. They are + keyword-only, so their defaults live in a plain writable dict on the function + object, which is the seam used here. +2. The count reset lives inside the method body, where no keyword default can + reach it. So this module reads the shipped source of + ``_stream_stdio_with_retries``, verifies its shape against the AST it expects, + inserts the one missing line, and rebinds the recompiled coroutine. + +Seam 2 is the fragile one and deserves a plain statement of that: it patches a +private async generator of a pinned dependency by recompiling its source. It is +structured to fail loudly and completely rather than partially. Every assumption +(the method exists, is undecorated, has no closure, assigns the delay exactly +twice with exactly one of those inside the read loop, assigns the count exactly +once outside it) is asserted before anything is written, and any surprise leaves +Modal's own method installed untouched. A modal upgrade that reshapes the loop +therefore degrades to seam 1 alone with a warning in the job log, never to a +silently mismatched hybrid. + +Loaded by being on ``PYTHONPATH`` when vero spawns harbor (see +``vero.harbor.cli``). Inert unless vero sets the environment variables below, so +importing it in any other context does nothing. +""" + +import os +import sys + +_DELAY_ENV = "VERO_MODAL_STREAM_RETRY_DELAY_SECS" +_FACTOR_ENV = "VERO_MODAL_STREAM_RETRY_FACTOR" +_RETRIES_ENV = "VERO_MODAL_STREAM_MAX_RETRIES" + +_SETTINGS = ( + ("stream_stdio_retry_delay_secs", _DELAY_ENV, float), + ("stream_stdio_retry_delay_factor", _FACTOR_ENV, float), + ("stream_stdio_max_retries", _RETRIES_ENV, int), +) + +# Names this module has to recognise in modal's source to place the reset. Kept +# as constants because they are simultaneously the thing asserted and the thing +# written: if a modal upgrade renames either local, the assertions below fail and +# nothing is written, rather than an edit landing next to a stale name. +_METHOD_NAME = "_stream_stdio_with_retries" +_COUNT_LOCAL = "num_retries_remaining" +_COUNT_ATTRIBUTE = "stream_stdio_max_retries" +_DELAY_LOCAL = "delay_secs" +_DELAY_ATTRIBUTE = "stream_stdio_retry_delay_secs" + + +def _warn(message: str) -> None: + """Report to stderr, which harbor captures into the run's job log. + + A patch of a private module in a pinned dependency has to be loud when it + stops applying. Silence would read as "the budget is per-outage" while the + run is back on a lifetime budget it can exhaust in its first hour. + """ + + print(f"vero: modal stream patch: {message}", file=sys.stderr, flush=True) + + +def _chain_to_shadowed_sitecustomize() -> None: + """Run any ``sitecustomize`` this module is shadowing on ``sys.path``. + + Python imports exactly one module by that name. Prepending our directory to + PYTHONPATH would otherwise silently disable an interpreter's own, so find the + next one along the path and execute it first. + """ + + import importlib.util + + here = os.path.dirname(os.path.abspath(__file__)) + for entry in sys.path: + try: + if not entry or os.path.abspath(entry) == here: + continue + candidate = os.path.join(entry, "sitecustomize.py") + if not os.path.isfile(candidate): + continue + spec = importlib.util.spec_from_file_location("_vero_shadowed", candidate) + if spec is None or spec.loader is None: + continue + spec.loader.exec_module(importlib.util.module_from_spec(spec)) + return + except Exception as error: # never let chaining break the run + _warn(f"could not chain to {candidate}: {type(error).__name__}: {error}") + return + + +def _assigns_attribute(node, local: str, attribute: str) -> bool: + """True for exactly `` = self.``, nothing looser. + + Matched on the AST rather than on text so that reformatting, a reworded + comment, or a changed indent cannot silently move where the reset lands. + """ + + import ast + + return ( + isinstance(node, ast.Assign) + and len(node.targets) == 1 + and isinstance(node.targets[0], ast.Name) + and node.targets[0].id == local + and isinstance(node.value, ast.Attribute) + and node.value.attr == attribute + and isinstance(node.value.value, ast.Name) + and node.value.value.id == "self" + ) + + +def _within_read_loop(tree) -> set: + """Ids of nodes under an ``async for``, i.e. modal's chunk-consuming loop. + + This is how the initialisation of a local is told apart from its per-chunk + reset without depending on line numbers or on the comment above it. + """ + + import ast + + inside = set() + for node in ast.walk(tree): + if isinstance(node, ast.AsyncFor): + inside.update(id(child) for child in ast.walk(node)) + return inside + + +def _rewrite_source(source: str): + """Insert the count reset beside modal's delay reset, or return None. + + None means an assumption failed and the caller must leave modal's own method + in place. Every check here guards a shape this module would otherwise be + editing blind. + """ + + import ast + + tree = ast.parse(source) + if len(tree.body) != 1 or not isinstance(tree.body[0], ast.AsyncFunctionDef): + _warn(f"{_METHOD_NAME} is not a lone async def; budget stays lifetime-scoped") + return None + function = tree.body[0] + if function.name != _METHOD_NAME or function.decorator_list: + # A decorator would be dropped by recompiling the def alone, which would + # quietly change behaviour rather than fail. Refuse instead. + _warn(f"{_METHOD_NAME} is decorated or renamed; budget stays lifetime-scoped") + return None + + in_loop = _within_read_loop(tree) + nodes = list(ast.walk(tree)) + delays = [n for n in nodes if _assigns_attribute(n, _DELAY_LOCAL, _DELAY_ATTRIBUTE)] + counts = [n for n in nodes if _assigns_attribute(n, _COUNT_LOCAL, _COUNT_ATTRIBUTE)] + resets = [n for n in delays if id(n) in in_loop] + # modal 1.5.3: the delay is assigned twice (once above `while True`, once per + # successful chunk) and the count exactly once, above the loop. Anything else + # means the loop was reshaped and the premise of this patch no longer holds. + if ( + len(delays) != 2 + or len(resets) != 1 + or len(counts) != 1 + or id(counts[0]) in in_loop + ): + _warn( + f"modal's retry loop changed shape ({len(delays)} delay assignments, " + f"{len(resets)} of them per-chunk, {len(counts)} count assignments); " + "budget stays lifetime-scoped" + ) + return None + + reset = resets[0] + lines = source.splitlines(keepends=True) + lines.insert( + reset.end_lineno, + " " * reset.col_offset + + f"{_COUNT_LOCAL} = self.{_COUNT_ATTRIBUTE}" + + " # vero: per-outage budget, see vero/harbor/_stream_patch\n", + ) + patched = "".join(lines) + + # Re-read what was actually produced. Cheap, and it is the difference between + # "we intended to add a reset" and "a reset exists inside the read loop". + verify = ast.parse(patched) + verify_loop = _within_read_loop(verify) + landed = [ + n + for n in ast.walk(verify) + if _assigns_attribute(n, _COUNT_LOCAL, _COUNT_ATTRIBUTE) and id(n) in verify_loop + ] + if len(landed) != 1: + _warn("rewrite did not land the count reset in the read loop; not applying") + return None + return patched + + +def _make_budget_per_outage(client) -> bool: + """Rebind ``_stream_stdio_with_retries`` so a good chunk restores the count. + + Recompiled against modal's own module globals (not a copy) so the private + names the body reads -- ``sr_pb2``, ``RETRYABLE_GRPC_STATUS_CODES``, + ``ExecTimeoutError``, the module ``logger`` -- resolve to the live objects, + and so a later modal-side rebinding of any of them is still seen. + """ + + import inspect + import textwrap + + original = client.__dict__.get(_METHOD_NAME) + if original is None or not inspect.isasyncgenfunction(original): + _warn( + f"{_METHOD_NAME} absent or not an async generator; " + "budget stays lifetime-scoped" + ) + return False + if original.__closure__ is not None: + # A closure cell (``super()``, ``__class__``) cannot be rebuilt by + # recompiling the source standalone; the result would raise at call time. + _warn(f"{_METHOD_NAME} closes over cells; budget stays lifetime-scoped") + return False + + try: + source = textwrap.dedent(inspect.getsource(original)) + except (OSError, TypeError) as error: + _warn( + f"{_METHOD_NAME} source unavailable ({error}); budget stays lifetime-scoped" + ) + return False + + patched_source = _rewrite_source(source) + if patched_source is None: + return False + + globals_ = original.__globals__ + if _METHOD_NAME in globals_: + # Executing the def would clobber a module-level name of the same spelling. + _warn(f"{_METHOD_NAME} shadows a module global; budget stays lifetime-scoped") + return False + try: + code = compile(patched_source, "", "exec") + exec(code, globals_) + patched = globals_.pop(_METHOD_NAME) + except Exception as error: + globals_.pop(_METHOD_NAME, None) + _warn( + f"could not recompile {_METHOD_NAME} " + f"({type(error).__name__}: {error}); budget stays lifetime-scoped" + ) + return False + + if not inspect.isasyncgenfunction(patched): + _warn(f"recompiled {_METHOD_NAME} is not an async generator; not applying") + return False + setattr(client, _METHOD_NAME, patched) + return True + + +def _apply() -> None: + requested = { + name: os.environ[env] + for name, env, _ in _SETTINGS + if os.environ.get(env) not in (None, "") + } + if not requested: + return # vero did not ask; stay inert + + try: + from modal._utils.task_command_router_client import TaskCommandRouterClient + except Exception as error: + _warn(f"modal client not importable, defaults unchanged: {error}") + return + + defaults = getattr(TaskCommandRouterClient.__init__, "__kwdefaults__", None) + if not isinstance(defaults, dict): + _warn("client __init__ has no keyword defaults; modal API changed") + return + + missing = [name for name, _, _ in _SETTINGS if name not in defaults] + if missing: + _warn(f"modal API changed, absent knobs {missing}; defaults unchanged") + return + + # Parse every value before writing any, because the three are only meaningful + # as a set. Committing the valid ones would pair a raised retry count with + # Modal's exponential default, and that combination does not fail closed: the + # sleeps double from the delay, so ~60 retries means sleeps longer than the + # run. A bad factor would turn the crash this fixes into a hang. + applied = {} + for name, env, cast in _SETTINGS: + raw = os.environ.get(env) + if raw in (None, ""): + continue + try: + applied[name] = cast(raw) + except ValueError: + _warn( + f"{env}={raw!r} is not a valid {cast.__name__}; " + "leaving ALL reconnect settings at modal's defaults" + ) + return + + # Parseable is not the same as safe. A factor above 1.0 makes each sleep grow + # from the one before, so a raised retry count stops being a wider window and + # becomes an unbounded one: at the shipped factor of 2, the sixtieth sleep is + # 2 * 2**59 seconds. That does not fail closed. It converts the crash this + # module exists to prevent into a hang, which is strictly harder to diagnose + # because the run neither finishes nor reports an error. + # + # Rejected rather than clamped. The whole premise here is that a long-lived + # stream needs PACING, and a flat delay is the only shape whose worst case is + # readable off the two numbers. Silently rewriting a caller's explicit choice + # would leave the log agreeing with a setting that is not in force. + factor = applied.get("stream_stdio_retry_delay_factor") + if factor is not None and factor > 1.0: + _warn( + f"{_FACTOR_ENV}={factor} grows every sleep from the last, so the " + "budget is unbounded rather than paced; leaving ALL reconnect " + "settings at modal's defaults. Use 1.0 for a flat delay." + ) + return + + defaults.update(applied) + + # After the knobs, and deliberately not gated on each other. If the rewrite + # fails, the knobs alone still leave a strictly wider budget than modal ships + # (a flat 2s beats a 0.01s doubling spent in 10.23s), so withholding them on + # that failure would be worse for the run, not safer. They are independent + # improvements, and the reported scope below says which one is in force. + per_outage = _make_budget_per_outage(TaskCommandRouterClient) + + if applied: + delay = applied.get("stream_stdio_retry_delay_secs") + factor = applied.get("stream_stdio_retry_delay_factor") + tries = applied.get("stream_stdio_max_retries") + scope = "per outage" if per_outage else "per stream, lifetime" + window = ( + f", ~{delay * tries:.0f}s tolerated {scope}" + if None not in (delay, tries) and factor == 1.0 + else f", budget is {scope}" + ) + _warn(f"applied {applied}{window}") + + +_chain_to_shadowed_sitecustomize() +_apply() diff --git a/vero/src/vero/harbor/cli.py b/vero/src/vero/harbor/cli.py index 90562c56..72253b8d 100644 --- a/vero/src/vero/harbor/cli.py +++ b/vero/src/vero/harbor/cli.py @@ -237,6 +237,71 @@ def _load_env_file(path: Path) -> dict[str, str]: # override and calls the provider's public endpoint instead of the gateway. _LITELLM_AGENTS = frozenset({"mini-swe-agent", "swe-agent"}) +# How long a dropped Modal stdio stream keeps trying to reconnect before the +# trial dies, and how often. These pace a budget that `_stream_patch` first has +# to make per-outage; read that module's docstring before touching them. +# +# Harbor reads a whole agent phase through one Modal stdio stream, and Modal +# budgets reconnects for the LIFE of that stream: `stream_stdio_max_retries` is +# set once above the read loop and never replenished, because a successful chunk +# resets only the backoff delay (modal 1.5.3, +# `_utils/task_command_router_client.py:769` and :801-802). A measured optimizer +# phase held one stream open for 3h19m, so unrelated blips hours apart draw down +# the same counter and the drop that kills the trial is rarely the one that +# earned it. Shipped as 10 attempts at 0.01s doubling, the whole budget is also +# spent in 10.23 seconds of continuous outage. Two cells died that way on +# 2026-07-31 and 2026-08-01. +# +# Raising the count was the first fix and it was the wrong one: any finite +# lifetime cap is the wrong shape for an hours-long stream, and with a factor of +# 2 the sleeps outgrow the run by roughly the seventeenth attempt, so a higher +# count alone converts a crash into a hang. `_stream_patch` now makes the count +# reset per successful chunk, which is what the delay already did. The flat +# factor stays: it keeps every gap short and makes the window below a product. +# +# MODAL_STREAM_RECONNECT_WINDOW_SECONDS is therefore how long ONE outage may last +# before the trial dies. It is capped by how much stdout the worker keeps +# available for a reconnect at a byte offset. Past that the reconnect does not +# fail cleanly, it resumes past the retained span, so output goes missing instead +# of the run dying. Waiting longer would be worse than dying, which is what +# bounds this and not the length of a typical outage. +# +# The retained span is measured in BYTES, not seconds, so the safe window in +# seconds scales inversely with how chatty the harness is. Probed against live +# sandboxes: 6 KB over 150s and 240 KB over 120s both reconnected contiguously, +# 3 MB over 30s came back empty. The ceiling is therefore a byte count somewhere +# above 240 KB, and how many seconds that buys depends entirely on output rate. +# +# A measured opencode optimizer transcript runs ~350 B/s, so 600s of outage is +# ~210 KB, still inside the 240 KB that was verified to reconnect contiguously. +# That is what sets this value: the widest window that stays within the +# measured-safe envelope at the measured rate, not a guess at how long drops last. +# +# This was 120s (~42 KB) first, about 6x more conservative than the evidence +# required. The phase it protects runs for hours, so a two-minute ceiling let one +# blip end work that had already cost hours and dollars, which is the wrong price +# for 42 KB of exposure. Widening it does not weaken the byte bound, it stops +# leaving most of the bound unused. +# +# Lower this for a harness chattier than the ~350 B/s measured, and re-measure +# before raising it further: the rate, not the elapsed seconds, is what overruns +# the buffer. Nothing here measures how long a real drop lasts, and the +# outer-trial retry remains the backstop past this window. +MODAL_STREAM_RECONNECT_DELAY_SECONDS = 2.0 +MODAL_STREAM_RECONNECT_WINDOW_SECONDS = 600 +MODAL_STREAM_RECONNECT_FACTOR = 1.0 + +# Directory holding the `sitecustomize` that applies the above inside the harbor +# subprocess, and that makes the budget per-outage. Modal exposes these three +# only as constructor keywords that `TaskCommandRouterClient._connect` never +# forwards, and `modal/config.py` has no entry for them, so there is no supported +# path: not an argument, not an environment variable. PYTHONPATH plus +# `sitecustomize` is the seam that reaches a dependency's defaults without +# vendoring it. The per-outage reset is not reachable that way at all, since it +# lives inside the method body, so that half recompiles modal's own source after +# checking its shape; the module's docstring is explicit about the cost. +_STREAM_PATCH_DIRECTORY = Path(__file__).parent / "_stream_patch" + def _litellm_base_url_args(agent: str, task: Path) -> list[str]: """Give litellm-based harnesses the gateway URL under the name they read. @@ -464,6 +529,36 @@ def _agent_environment_blanks(task: Path) -> list[str]: return arguments +def _modal_stream_patch_environment(base: dict[str, str]) -> dict[str, str]: + """Put the reconnect-budget `sitecustomize` on the harbor subprocess's path. + + Prepended to any inherited PYTHONPATH rather than replacing it, and the + `sitecustomize` chains to whichever module it shadows, so a caller who + already relies on one keeps it. + + A caller who sets the three values explicitly keeps them: this fills in the + vero defaults and never overrides an explicit choice, which is what makes the + window tunable per run without a code change. + """ + + patched = {"PYTHONPATH": str(_STREAM_PATCH_DIRECTORY)} + inherited = base.get("PYTHONPATH") + if inherited: + patched["PYTHONPATH"] = os.pathsep.join([patched["PYTHONPATH"], inherited]) + + retries = int( + MODAL_STREAM_RECONNECT_WINDOW_SECONDS / MODAL_STREAM_RECONNECT_DELAY_SECONDS + ) + for name, value in ( + ("VERO_MODAL_STREAM_RETRY_DELAY_SECS", MODAL_STREAM_RECONNECT_DELAY_SECONDS), + ("VERO_MODAL_STREAM_RETRY_FACTOR", MODAL_STREAM_RECONNECT_FACTOR), + ("VERO_MODAL_STREAM_MAX_RETRIES", retries), + ): + if not base.get(name): + patched[name] = str(value) + return patched + + def _compiled_run_environment( task: Path, overrides: dict[str, str] | None = None ) -> dict[str, str]: @@ -476,6 +571,7 @@ def _compiled_run_environment( environment = os.environ.copy() if overrides: environment.update(overrides) + environment.update(_modal_stream_patch_environment(environment)) path = task / "environment/gateway/launch.json" if not path.exists(): return environment diff --git a/vero/tests/test_v05_modal_stream_patch.py b/vero/tests/test_v05_modal_stream_patch.py new file mode 100644 index 00000000..374d79f0 --- /dev/null +++ b/vero/tests/test_v05_modal_stream_patch.py @@ -0,0 +1,480 @@ +"""The Modal stdio reconnect budget vero applies to its harbor subprocess. + +Modal budgets stdio reconnects for the LIFE of a stream: the count is set once +above the read loop and a successful chunk resets only the backoff delay, so an +hours-long harbor phase spends the budget on blips unrelated to the drop that +eventually kills it. vero reaches into the subprocess with a `sitecustomize` on +PYTHONPATH and fixes two things there: the three numeric knobs (constructor +keywords that `TaskCommandRouterClient._connect` never forwards and +`modal/config.py` never exposes), and the missing per-chunk reset of the count, +which lives inside the method body and so needs modal's own source recompiled. + +These tests pin both seams, because a silent no-op in either reads exactly like a +budget that survives the run. +""" + +from __future__ import annotations + +import os +import runpy +import subprocess +import sys +import tempfile +import textwrap + +from vero.harbor import cli as harbor_cli + +PATCH_DIRECTORY = harbor_cli._STREAM_PATCH_DIRECTORY +KNOBS = ( + "stream_stdio_retry_delay_secs", + "stream_stdio_retry_delay_factor", + "stream_stdio_max_retries", +) + + +def _run_patch(environment: dict[str, str], modal_source: str) -> tuple[str, str]: + """Execute the sitecustomize against a stand-in modal package. + + A stand-in rather than the real client: the point is the seam (keyword + defaults on `__init__`), and pinning it against a fake keeps the test honest + when modal is absent from the test environment. The end-to-end check against + the installed modal lives in `test_patch_applies_to_the_real_modal_client`. + """ + + # Written to a file rather than passed with `-c`, because half the patch + # recompiles modal's own source and `inspect.getsource` needs a real file to + # read it from. A `-c` stand-in would decline the rewrite for a reason that + # has nothing to do with the seam under test. (The same decline is the right + # behaviour against a sourceless modal install, and is asserted separately.) + directory = tempfile.mkdtemp(prefix="vero-stream-patch-") + script = os.path.join(directory, "drive_fake_modal.py") + with open(script, "w", encoding="utf-8") as handle: + handle.write(textwrap.dedent(modal_source)) + process = subprocess.run( + [sys.executable, script], + env={ + **os.environ, + "PYTHONPATH": os.pathsep.join( + [str(PATCH_DIRECTORY), os.environ.get("PYTHONPATH", "")] + ).strip(os.pathsep), + **environment, + }, + capture_output=True, + text=True, + ) + return process.stdout, process.stderr + + +# A miniature of modal 1.5.3's `_stream_stdio_with_retries`, kept faithful in the +# ways the patch depends on: the delay assigned twice (once above `while True`, +# once per successful chunk) and the count assigned once, above the loop. Both +# seams are pinned against this rather than against the real client so the tests +# still run where modal is not installed; `test_patch_applies_to_the_real_modal_ +# client` is what checks the shape has not drifted from the dependency. +FAKE_MODAL_MODULE = """ + import asyncio, sys, types + package = types.ModuleType("modal") + utils = types.ModuleType("modal._utils") + module = types.ModuleType("modal._utils.task_command_router_client") + + class TaskCommandRouterClient: + def __init__( + self, + server_client, + *, + stream_stdio_retry_delay_secs: float = 0.01, + stream_stdio_retry_delay_factor: float = 2, + stream_stdio_max_retries: int = 10, + ) -> None: + self.stream_stdio_retry_delay_secs = stream_stdio_retry_delay_secs + self.stream_stdio_retry_delay_factor = stream_stdio_retry_delay_factor + self.stream_stdio_max_retries = stream_stdio_max_retries + + def _get_metadata(self): + return {} + + async def _stream_stdio_with_retries( + self, *, stub_method, request_factory, deadline_label, deadline=None + ): + offset = 0 + delay_secs = self.stream_stdio_retry_delay_secs + delay_factor = self.stream_stdio_retry_delay_factor + num_retries_remaining = self.stream_stdio_max_retries + + async def sleep_and_update(e): + nonlocal delay_secs, num_retries_remaining + await asyncio.sleep(delay_secs) + delay_secs *= delay_factor + num_retries_remaining -= 1 + + while True: + try: + stream = stub_method.open(timeout=None, metadata=self._get_metadata()) + async with stream as s: + await s.send_message(request_factory(offset), end=True) + async for item in s: + # Reset retry backoff after any successful chunk. + delay_secs = self.stream_stdio_retry_delay_secs + offset += len(item.data) + yield item + return + except OSError as e: + if num_retries_remaining > 0: + await sleep_and_update(e) + else: + raise + + module.TaskCommandRouterClient = TaskCommandRouterClient + sys.modules["modal"] = package + sys.modules["modal._utils"] = utils + sys.modules["modal._utils.task_command_router_client"] = module +""" + +FAKE_MODAL = ( + FAKE_MODAL_MODULE + + """ + import sitecustomize + sitecustomize._apply() + print({k: v for k, v in TaskCommandRouterClient.__init__.__kwdefaults__.items()}) +""" +) + +# Scripted outage sequence: three attempts that deliver one chunk and then drop, +# then one that delivers a chunk and ends. Every outage is separated by a +# success, so a lifetime budget of 2 is exhausted by the third drop while a +# per-outage budget of 2 is never below 2 when a drop arrives. The client is +# built with no keyword arguments on purpose: the budget it runs on has to come +# from the patched `__kwdefaults__`, the way `_connect` builds it. +FAKE_MODAL_DRIVEN = ( + FAKE_MODAL_MODULE + + """ + import json, sitecustomize + sitecustomize._apply() + + ATTEMPTS = 4 + + class Chunk: + def __init__(self, data): + self.data = data + + class Attempt: + def __init__(self, index, drops): + self.index, self.drops = index, drops + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def send_message(self, request, end=True): + return None + + def __aiter__(self): + async def chunks(): + yield Chunk(b"chunk\\n") + if self.drops: + raise OSError("simulated drop %d" % self.index) + return chunks() + + class Stub: + def __init__(self): + self.opened = 0 + + def open(self, timeout=None, metadata=None): + self.opened += 1 + return Attempt(self.opened, drops=self.opened < ATTEMPTS) + + async def drive(): + client = TaskCommandRouterClient(None) + delivered, outcome = 0, "completed" + stream = client._stream_stdio_with_retries( + stub_method=Stub(), request_factory=lambda offset: None, + deadline_label="test", + ) + try: + async for item in stream: + delivered += 1 + except OSError as error: + outcome = "died: %s" % error + return {"delivered": delivered, "outcome": outcome, "attempts": ATTEMPTS} + + print(json.dumps(asyncio.run(drive()))) +""" +) + + +def test_patch_widens_the_reconnect_budget_when_vero_asks() -> None: + stdout, stderr = _run_patch( + { + "VERO_MODAL_STREAM_RETRY_DELAY_SECS": "2.0", + "VERO_MODAL_STREAM_RETRY_FACTOR": "1.0", + "VERO_MODAL_STREAM_MAX_RETRIES": "60", + }, + FAKE_MODAL, + ) + defaults = eval(stdout.strip().splitlines()[-1]) + assert defaults["stream_stdio_retry_delay_secs"] == 2.0 + assert defaults["stream_stdio_retry_delay_factor"] == 1.0 + assert defaults["stream_stdio_max_retries"] == 60 + # A flat factor is what makes the tolerated outage a simple product; a + # doubling one would outgrow the run instead of pacing it. "per outage" is + # load-bearing in this message: the same numbers spent per stream are the bug + # this patch was rewritten to fix, so the job log has to name the scope. + assert "120s tolerated per outage" in stderr + + +def _drive(environment: dict[str, str], source: str = FAKE_MODAL_DRIVEN) -> tuple: + import json + + stdout, stderr = _run_patch(environment, source) + return json.loads(stdout.strip().splitlines()[-1]), stderr + + +PER_OUTAGE_ENVIRONMENT = { + "VERO_MODAL_STREAM_RETRY_DELAY_SECS": "0.0", + "VERO_MODAL_STREAM_RETRY_FACTOR": "1.0", + "VERO_MODAL_STREAM_MAX_RETRIES": "2", +} + + +def test_a_lifetime_budget_dies_on_outages_it_has_already_survived() -> None: + """The defect, stated as a test: this is modal 1.5.3 without the rewrite. + + Three drops, each separated by a chunk that arrived fine, against a budget of + two. The count is set once above the read loop and only the delay is reset + per chunk, so the two are spent by the second drop and the third is fatal + even though the stream has been proving itself healthy in between. Scaled up, + that is a 3h19m optimizer phase dying to a blip because of blips hours + earlier. + """ + + result, _ = _drive(PER_OUTAGE_ENVIRONMENT, _reshaped_loop()) + assert result["delivered"] == 3 + assert "died: simulated drop 3" in result["outcome"] + + +def test_patch_makes_the_retry_budget_per_outage() -> None: + """The fix: a successful chunk restores the count, not just the delay. + + Same script, same budget of two, and now every drop is met with a full + budget because the stream delivered a chunk since the last one. What the + patch buys is not more retries, it is retries that belong to the outage in + front of them. + """ + + result, stderr = _drive(PER_OUTAGE_ENVIRONMENT) + assert result["delivered"] == result["attempts"] == 4 + assert result["outcome"] == "completed" + assert "per outage" in stderr + + +def _reshaped_loop() -> str: + """`FAKE_MODAL_DRIVEN` with the per-chunk delay reset written differently. + + Same behaviour, different AST, which is exactly the modal upgrade this patch + has to survive: the rewrite must decline rather than guess where the count + reset belongs. The last occurrence is the in-loop one. + """ + + head, tail = FAKE_MODAL_DRIVEN.rsplit( + "delay_secs = self.stream_stdio_retry_delay_secs", 1 + ) + return head + "delay_secs = max(0.0, self.stream_stdio_retry_delay_secs)" + tail + + +def test_patch_declines_a_reshaped_retry_loop_and_says_so() -> None: + """Recompiling a pinned dependency's private coroutine has to fail closed. + + A reshaped loop leaves modal's own method installed and untouched (the run + dies at three chunks, exactly as unpatched), warns loudly, and still applies + the numeric knobs, since a flat 2s budget is strictly wider than the 0.01s + doubling modal ships and withholding it would punish the run for the + rewrite's failure rather than protect it. + """ + + result, stderr = _drive(PER_OUTAGE_ENVIRONMENT, _reshaped_loop()) + assert result["delivered"] == 3 + assert "changed shape" in stderr + assert "budget stays lifetime-scoped" in stderr + # The knobs landed even though the rewrite did not, and the message says + # which scope is actually in force rather than implying the good one. + assert "per stream, lifetime" in stderr + + +def test_patch_is_inert_when_vero_does_not_ask() -> None: + """Importable anywhere without changing behaviour, so it cannot surprise.""" + + stdout, stderr = _run_patch({}, FAKE_MODAL) + defaults = eval(stdout.strip().splitlines()[-1]) + assert defaults["stream_stdio_max_retries"] == 10 + assert stderr == "" + + +def test_patch_is_loud_when_modal_renames_the_knobs() -> None: + """The failure that matters: a modal upgrade turning this into a no-op. + + Silence would read as "the budget is wide" while the run is back on ten + seconds of tolerance, so an absent knob has to reach the job log. + """ + + renamed = FAKE_MODAL.replace("stream_stdio_max_retries", "stream_stdio_max_attempts") + _, stderr = _run_patch({"VERO_MODAL_STREAM_MAX_RETRIES": "60"}, renamed) + assert "modal API changed" in stderr + assert "stream_stdio_max_retries" in stderr + + +def test_environment_puts_the_patch_on_the_subprocess_path() -> None: + environment = harbor_cli._modal_stream_patch_environment({}) + assert environment["PYTHONPATH"].split(os.pathsep)[0] == str(PATCH_DIRECTORY) + assert (PATCH_DIRECTORY / "sitecustomize.py").is_file() + + retries = int( + harbor_cli.MODAL_STREAM_RECONNECT_WINDOW_SECONDS + / harbor_cli.MODAL_STREAM_RECONNECT_DELAY_SECONDS + ) + assert environment["VERO_MODAL_STREAM_MAX_RETRIES"] == str(retries) + + +def test_environment_preserves_an_inherited_pythonpath() -> None: + """Prepend, never replace: a caller's own PYTHONPATH has to survive.""" + + environment = harbor_cli._modal_stream_patch_environment({"PYTHONPATH": "/opt/mine"}) + assert environment["PYTHONPATH"] == os.pathsep.join( + [str(PATCH_DIRECTORY), "/opt/mine"] + ) + + +def test_environment_defers_to_an_explicit_window() -> None: + """A run can widen or narrow the window without a code change.""" + + environment = harbor_cli._modal_stream_patch_environment( + {"VERO_MODAL_STREAM_MAX_RETRIES": "5"} + ) + assert "VERO_MODAL_STREAM_MAX_RETRIES" not in environment + + +def test_patch_applies_to_the_real_modal_client() -> None: + """End to end against whichever modal is installed, or skipped if absent. + + The fake pins the seam's shape; only this pins that the shape still matches + the dependency vero actually patches. + """ + + import pytest + + pytest.importorskip("modal._utils.task_command_router_client") + stdout, _ = _run_patch( + { + "VERO_MODAL_STREAM_RETRY_DELAY_SECS": "2.0", + "VERO_MODAL_STREAM_RETRY_FACTOR": "1.0", + "VERO_MODAL_STREAM_MAX_RETRIES": "60", + }, + """ + from modal._utils.task_command_router_client import TaskCommandRouterClient + state = {k: v for k, v in TaskCommandRouterClient.__init__.__kwdefaults__.items() + if 'stdio' in k} + state['recompiled'] = ( + TaskCommandRouterClient._stream_stdio_with_retries.__code__.co_filename + ) + print(state) + """, + ) + defaults = eval(stdout.strip().splitlines()[-1]) + assert defaults["stream_stdio_max_retries"] == 60 + assert defaults["stream_stdio_retry_delay_secs"] == 2.0 + # The rewrite is the half that cannot be expressed as a keyword default, and + # the only proof it took is that the installed method now comes from vero's + # compile rather than modal's file. + assert defaults["recompiled"] == "" + + +def test_sitecustomize_chains_to_a_shadowed_module(tmp_path) -> None: + """Prepending our directory must not disable an interpreter's own. + + Python imports exactly one module named `sitecustomize`, so shadowing one + silently is a real risk of this mechanism rather than a hypothetical. + """ + + shadowed = tmp_path / "sitecustomize.py" + shadowed.write_text("import sys; print('shadowed ran', file=sys.stderr)\n") + process = subprocess.run( + [sys.executable, "-c", "pass"], + env={ + **os.environ, + "PYTHONPATH": os.pathsep.join([str(PATCH_DIRECTORY), str(tmp_path)]), + }, + capture_output=True, + text=True, + ) + assert "shadowed ran" in process.stderr + + +def test_module_is_importable_without_side_effects() -> None: + """`runpy` it directly: no modal, no environment, no exception.""" + + runpy.run_path(str(PATCH_DIRECTORY / "sitecustomize.py"), run_name="not_main") + + +def test_one_bad_value_leaves_every_setting_at_modal_defaults() -> None: + """All or nothing: the three are only meaningful together. + + Committing the valid ones would pair a raised retry count with modal's + exponential default, and that does not fail closed. Sleeps double from the + delay, so ~60 retries means sleeps longer than the run: the crash this fixes + becomes a hang, which is harder to diagnose than what it replaced. + """ + + stdout, stderr = _run_patch( + { + "VERO_MODAL_STREAM_RETRY_DELAY_SECS": "2.0", + "VERO_MODAL_STREAM_RETRY_FACTOR": "not-a-float", + "VERO_MODAL_STREAM_MAX_RETRIES": "60", + }, + FAKE_MODAL, + ) + defaults = eval(stdout.strip().splitlines()[-1]) + assert defaults["stream_stdio_max_retries"] == 10 + assert defaults["stream_stdio_retry_delay_secs"] == 0.01 + assert defaults["stream_stdio_retry_delay_factor"] == 2 + assert "leaving ALL reconnect settings" in stderr + + +def test_a_growing_factor_is_refused_rather_than_accepted() -> None: + """Parseable is not safe: a factor above 1.0 makes the budget unbounded. + + Greptile flagged this on PR #79. With modal's shipped factor of 2 and a + raised count, the sixtieth sleep is 2 * 2**59 seconds, so the "wider window" + is really no window at all. It does not fail closed either: the run neither + finishes nor errors, which is harder to diagnose than the crash this module + prevents. Refused as a set, so no half-applied policy survives. + """ + + stdout, stderr = _run_patch( + { + "VERO_MODAL_STREAM_RETRY_DELAY_SECS": "2.0", + "VERO_MODAL_STREAM_RETRY_FACTOR": "2", + "VERO_MODAL_STREAM_MAX_RETRIES": "60", + }, + FAKE_MODAL, + ) + defaults = eval(stdout.strip().splitlines()[-1]) + assert defaults["stream_stdio_max_retries"] == 10 + assert defaults["stream_stdio_retry_delay_secs"] == 0.01 + assert defaults["stream_stdio_retry_delay_factor"] == 2 + assert "unbounded rather than paced" in stderr + assert "Use 1.0 for a flat delay" in stderr + + +def test_a_flat_factor_is_still_accepted() -> None: + """The guard must not reject the value the module actually ships.""" + + stdout, _ = _run_patch( + { + "VERO_MODAL_STREAM_RETRY_DELAY_SECS": "2.0", + "VERO_MODAL_STREAM_RETRY_FACTOR": "1.0", + "VERO_MODAL_STREAM_MAX_RETRIES": "60", + }, + FAKE_MODAL, + ) + assert eval(stdout.strip().splitlines()[-1])["stream_stdio_max_retries"] == 60