"""Producer-side byte budgets for everything a cell can emit. Three of the worker's output paths were bounded only when the response frame was assembled: kernel stderr, the formatted traceback, and — on the R side — the error string. `_cap` produced exactly the right string and exactly the wrong allocation, because by the time it ran the whole payload was already in worker RAM. Measured on the tree before the fix, with the same code these tests drive: * a cell doing 200 x `sys.stderr.write('x' * 1_000_000)` peaked at **452 MB** of traced allocation to retain 1 MB of it; * formatting a sixty-deep exception chain whose links share one 1 MB message peaked at **122 MB** to produce 61 MB of text that was then cut to 1 MB. Both returned the correct, capped string. That is why nothing caught this, and it is why every assertion below is about ALLOCATION rather than about the returned length: `tracemalloc` peak is the observable, and it does not depend on timing, machine speed, or when a thread happens to be scheduled. """ from __future__ import annotations import json import os import re import subprocess import sys import threading import time import tracemalloc from pathlib import Path import pytest from openai4s.kernel import sink_drain, worker from openai4s.kernel.background import BackgroundExecutor from openai4s.kernel.r_kernel import resolve_r_interpreter _R_WORKER = Path(__file__).parents[1] / "openai4s" / "kernel" / "r_worker.R" _REAL_R = resolve_r_interpreter() _MARKER_LEN = len(worker._TRUNCATION_MARKER) @pytest.fixture() def offline_worker(monkeypatch): """Run `_run_cell` in-process with no protocol channel underneath it. `_run_cell` is the production entry point — the same function `main()` calls for an `execute` frame — so this drives the real path. Only the two things that need a live subprocess around them are stubbed: the frame writer (there is no protocol fd here) and the lazy `host` install. """ frames: list[dict] = [] monkeypatch.setattr(worker, "_write_frame", frames.append) monkeypatch.setitem(worker._NS, "host", object()) # Warm every import `_run_cell` performs (guards, provenance, linecache) # so a measured peak is the cell's output and not a first module import. worker._run_cell("1", "warm-up") frames.clear() return frames def test_a_cell_writing_200mb_to_stderr_does_not_allocate_200mb(offline_worker): """The defect, and the only assertion that can see it. `err_buf` was a plain `io.StringIO`, capped in the response builder — the same mistake `_StreamingStdout` was written to fix on the other stream, left standing on this one. The returned string was already correct, so the length assertions below pass either way; the peak is what changes, from 452 MB to roughly 3 MB. """ code = ( "import sys\n" "for _ in range(200):\n" " sys.stderr.write('x' * 1_000_000)\n" ) tracemalloc.start() try: tracemalloc.reset_peak() response = worker._run_cell(code, "cell-stderr") _current, peak = tracemalloc.get_traced_memory() finally: tracemalloc.stop() assert response["error"] is None assert len(response["stderr"]) == worker.MAX_OUTPUT + _MARKER_LEN assert response["stderr"].count("...(truncated at") == 1 assert peak < 24_000_000, ( f"{peak} bytes peak to retain {worker.MAX_OUTPUT} characters — " "stderr is being accumulated whole and capped afterwards" ) def _chained_failure(depth: int, message: str) -> None: """Raise from inside `except`, so every level chains onto the last.""" if depth == 0: raise ValueError(message) try: _chained_failure(depth - 1, message) except ValueError: raise ValueError(message) def test_a_large_traceback_is_bounded_while_it_is_formatted(): """Built outside the measurement, formatted inside it. The chain is deliberately made of ONE message object shared by sixty links, so the exception graph itself costs 1 MB and the peak measured below is the formatter's, not the fixture's. `traceback.format_exc()` joins all sixty renderings before anything caps the result (122 MB); pulling the generator and stopping at the cap never formats the links past the first (3 MB). """ message = "y" * 1_000_000 caught: BaseException | None = None try: _chained_failure(60, message) except ValueError as exc: caught = exc assert caught is not None tracemalloc.start() try: tracemalloc.reset_peak() text = worker._bounded_format_exc(caught) _current, peak = tracemalloc.get_traced_memory() finally: tracemalloc.stop() assert len(text) == worker.MAX_OUTPUT + _MARKER_LEN assert text.count("...(truncated at") == 1 assert peak < 16_000_000, ( f"{peak} bytes peak to keep {worker.MAX_OUTPUT} characters of " "traceback — the whole chain is being formatted before it is capped" ) def test_a_traceback_that_fits_carries_no_marker(): """The cap must not claim a cut that did not happen. Stopping the generator at the cap is the easy way to attach a marker to output that merely ended there, which is the same silent lie the buffers guard against from the other direction. """ try: 1 / 0 except ZeroDivisionError as exc: text = worker._bounded_format_exc(exc) assert "truncated" not in text assert text.rstrip().endswith("ZeroDivisionError: division by zero") def test_a_huge_exception_message_still_produces_a_response_frame(offline_worker): """The failure the late cap left behind, asserted on the wire size. An uncapped `error` pushed the whole response past `_MAX_FRAME_BYTES`, `_write_frame` refused it and sent a `log` instead, and `Kernel.execute` then blocked on an id that never arrived until the watchdog killed the kernel. The frame has to fit, with the line number intact rather than lost with the rest of the payload. """ response = worker._run_cell("raise ValueError('e' * 12_000_000)", "cell-error") assert response["error"].count("...(truncated at") == 1 assert len(response["error"]) == worker.MAX_OUTPUT + _MARKER_LEN assert response["trace"]["error_lineno"] == 1 encoded = json.dumps(response, ensure_ascii=False).encode("utf-8") assert len(encoded) <= worker._MAX_FRAME_BYTES def test_a_trapped_system_exit_message_is_bounded_too(offline_worker): """`SystemExit('x' * N)` never reaches the traceback formatter. It is trapped and reported as a sentence, and that sentence used to interpolate the message whole. One marker, not two: the response builder no longer caps a string that has already been cut. """ response = worker._run_cell("raise SystemExit('z' * 3_000_000)", "cell-exit") assert response["error"].startswith("SystemExit trapped (worker kept alive): ") assert response["error"].count("...(truncated at") == 1 assert len(response["error"]) < 1_100_000 def test_the_bounded_buffer_reports_the_length_it_was_handed(): """The returned counter is the producer's, never the retained one. A short return means "partial write" to every stdlib caller, so reporting the retained length would make a bounded stream look like a failing one and invite the caller to send the remainder again. """ buf = worker._BoundedBuffer() assert buf.write("a" * 10) == 10 assert not buf.truncated and not buf.full assert buf.write("b" * (worker.MAX_OUTPUT * 2)) == worker.MAX_OUTPUT * 2 assert buf.full and buf.truncated # Writes after the cap are still acknowledged in full, and still dropped. assert buf.write("c" * 5) == 5 captured = buf.captured() assert len(captured) == worker.MAX_OUTPUT + _MARKER_LEN assert captured.count("...(truncated at") == 1 assert captured.startswith("a" * 10 + "b") def test_a_buffer_that_was_never_cut_says_nothing(): buf = worker._BoundedBuffer() buf.write("x" * worker.MAX_OUTPUT) # exactly the cap, nothing dropped assert buf.full and not buf.truncated assert buf.captured() == "x" * worker.MAX_OUTPUT def test_stdout_is_still_streamed_and_bounded_through_the_shared_base(monkeypatch): """`_StreamingStdout` now inherits its retention bound; prove it kept it. Sharing the bound with stderr is the point — two near-identical writers is how the two streams drifted apart in the first place — but the streaming half is `_StreamingStdout`'s alone and must survive the move. """ frames: list[dict] = [] monkeypatch.setattr(worker, "_write_frame", frames.append) buf = worker._StreamingStdout("cell-1") assert buf.write("hello") == 5 assert buf.write("x" * (worker.MAX_OUTPUT * 2)) == worker.MAX_OUTPUT * 2 assert buf.write("later") == 5 texts = [f["text"] for f in frames if f["type"] == "stdout_chunk"] assert texts, "the buffer streamed nothing at all" assert max(len(t) for t in texts) <= worker._MAX_CHUNK_CHARS streamed = "".join(texts) assert streamed.count("...(truncated at") == 1 assert len(streamed) <= worker.MAX_OUTPUT + _MARKER_LEN # The streamed tail and the captured result agree about what happened. assert buf.captured().count("...(truncated at") == 1 assert len(buf.captured()) == worker.MAX_OUTPUT + _MARKER_LEN # --- background capacity ---------------------------------------------------- class _BlockedKernel: """A kernel whose cell runs until something stops it.""" def __init__(self) -> None: self.release = threading.Event() def execute(self, code, origin="agent", on_chunk=None): del code, origin, on_chunk self.release.wait(5) return {"stdout": "", "error": None} def interrupt(self) -> None: self.release.set() def kill_worker(self) -> None: self.release.set() def shutdown(self) -> None: self.release.set() def test_a_background_slot_is_claimed_before_the_kernel_is_created(): """The ordering IS the fix, so the factory asserts on it from inside. `launch()` used to check `_closed`, release the lock, spawn a real worker process, and register the job afterwards — so concurrent launches all passed the check together and all spawned, and any cap tested there would have been tested against processes that already existed. The refusal below has to cost nothing: no worker for the job that was turned away. """ created: list[_BlockedKernel] = [] registered_at_spawn: list[int] = [] def factory() -> _BlockedKernel: # At the instant a worker would be spawned, this job is already in the # registry — the slot is claimed, not merely intended. registered_at_spawn.append(len(executor.list_jobs())) kernel = _BlockedKernel() created.append(kernel) return kernel executor = BackgroundExecutor(kernel_factory=factory, dispatcher=None) try: for _ in range(BackgroundExecutor.MAX_ACTIVE_JOBS): executor.launch("hang()") assert registered_at_spawn == list( range(1, BackgroundExecutor.MAX_ACTIVE_JOBS + 1) ) assert len(created) == BackgroundExecutor.MAX_ACTIVE_JOBS with pytest.raises(RuntimeError, match="already running"): executor.launch("one too many") assert len(created) == BackgroundExecutor.MAX_ACTIVE_JOBS, ( "the rejected launch spawned a worker anyway — capacity is being " "checked after the process exists" ) finally: executor.shutdown(timeout_per_job=1.0) def test_a_spawn_failure_returns_the_slot_it_claimed(): """Claiming first means the claim has to be given back. Without the rollback the cap leaks one slot per failed spawn, and after `MAX_ACTIVE_JOBS` failures the executor refuses every launch on a machine that is running nothing at all — the error even changes, from the real cause to a capacity message that is false. """ def factory(): raise RuntimeError("no pids left") executor = BackgroundExecutor(kernel_factory=factory, dispatcher=None) for _ in range(BackgroundExecutor.MAX_ACTIVE_JOBS + 4): with pytest.raises(RuntimeError, match="no pids left"): executor.launch("boom()") assert executor.list_jobs() == [] # --- the R sibling ---------------------------------------------------------- def test_the_r_worker_bounds_are_derived_from_the_python_ones(): """Both workers answer the same manager, so both owe the same contract. The R frame cap is read as the arithmetic it is rather than as a literal: a hand-typed number that happens to match today is exactly how the python side drifted from `MAX_OUTPUT` before it was derived. Needs no R. """ source = _R_WORKER.read_text(encoding="utf-8") cap = re.search(r"\.oai4s_MAX_OUTPUT <- (\d+)L", source) assert cap, "the R worker no longer declares a stream cap" assert int(cap.group(1)) == worker.MAX_OUTPUT frame = re.search(r"\.oai4s_MAX_FRAME_BYTES <- ([^\n]+)", source) assert frame, "the R worker has no outbound frame bound" expression = frame.group(1).strip() expression = expression.replace(".oai4s_MAX_OUTPUT", str(worker.MAX_OUTPUT)) value = eval( # noqa: S307 - arithmetic lifted from the R source, no names expression.replace("L", ""), {"__builtins__": {}}, {} ) assert value == worker._MAX_FRAME_BYTES @pytest.mark.skipif(_REAL_R is None, reason="no Rscript resolvable on this machine") def test_the_r_error_message_is_capped_at_the_producer(): """Drive `.oai4s_cap_message` itself, lifted out of the shipped worker. Lifting keeps the code under test the code that ships; sourcing the whole file would start the protocol loop, and a hand-written stand-in would be the thing that drifts. The helper is what the error path calls before the message is pasted into a bigger string and escaped character by character. """ source = _R_WORKER.read_text(encoding="utf-8") match = re.search(r"\.oai4s_cap_message <- function.*?\n\}", source, re.S) assert match, "the helper this test drives is no longer in the worker" script = ( ".oai4s_MAX_OUTPUT <- 1000000L\n" + match.group(0) + "\nout <- .oai4s_cap_message(strrep('x', 5000000L))\n" "short <- .oai4s_cap_message('boom')\n" "cat(nchar(out, type='chars'), nchar(short, type='chars'), '\\n')\n" ) result = subprocess.run( [_REAL_R, "--vanilla", "-e", script], capture_output=True, text=True, timeout=120, ) assert result.returncode == 0, result.stderr[:400] capped, short = (int(n) for n in result.stdout.split()[-2:]) assert capped == 1_000_000 + len("\n...(truncated at 1000000 characters)") assert short == len("boom") # nothing to cut, nothing said @pytest.mark.skipif(_REAL_R is None, reason="no Rscript resolvable on this machine") def test_real_r_caps_a_giant_error_instead_of_wiring_it(tmp_path): """End to end: `stop()` with a 50 MB message must not become a 50 MB field. This is the assertion that fails on the unfixed worker — the error string was the one field in the R response with no cap on it at all, and it went through `.oai4s_esc` a character at a time on its way to the pipe. """ from openai4s.kernel.r_kernel import spawn_r_kernel kernel = spawn_r_kernel(cwd=str(tmp_path), rscript=_REAL_R) try: result = kernel.execute("stop(strrep('x', 50000000L))") assert result["error"] is not None # Bounded is the property. The marker is asserted only when OUR cap is # what bit: R truncates `conditionMessage` itself (measured here: a 50 MB # `stop()` arrives as ~8 KB, so `.oai4s_cap_message` returns it # untouched), and which of the two limits applies depends on the R build. # Requiring the marker unconditionally would make this test assert the # local R version rather than the fix. assert len(result["error"]) <= 1_000_000 + 80 if len(result["error"]) > 1_000_000: assert result["error"].count("...(truncated at") == 1 # A short error is still reported verbatim. small = kernel.execute("stop('nope')") assert "nope" in small["error"] and "truncated" not in small["error"] finally: kernel.shutdown() # --- the host's drain, which is where the R streams are now bounded --------- def test_the_host_sink_cap_is_the_python_workers_cap(): """Restating a number is how the R side drifted from `MAX_OUTPUT` before. `sink_drain` cannot import `worker`: worker.py is a subprocess entry point that performs an fd swap when it runs, and the daemon has no reason to load it. So the constant is restated there and pinned here, the same way the R source's own caps are pinned above. """ assert sink_drain.CAP_BYTES == worker.MAX_OUTPUT def test_the_host_drain_counts_every_byte_it_declines_to_keep(): """Exact accounting is an acceptance criterion, not a nicety. The shell drainer this replaces (`sink()` into `pipe("head -c CAP > f; cat > /dev/null")`) failed it: `head -c` over-read the pipe it was bounding, so a 300,200-byte stream came out as 1,000 retained plus 298,976 counted -- 224 bytes unaccounted for, and no way to tell from the result whether they had been written or dropped. Driven with a real writer process on the other end of the fifo, because what is under test is the reader's behaviour against a producer it cannot pace, and an in-process writer would be paced by the GIL. """ directory = sink_drain.SinkDirectory() try: capture = directory.open(cap=1000) writer = subprocess.Popen( [ sys.executable, "-c", "import sys\n" "out = open(sys.argv[1], 'wb')\n" "out.write(b'a' * 300_200)\n" "out.close()\n" "err = open(sys.argv[2], 'wb')\n" "err.write(b'e' * 7)\n" "err.close()\n", capture.out_path, capture.err_path, ] ) assert writer.wait(timeout=30) == 0 stdout, stderr = capture.finish() counters = capture.counters() finally: directory.close() assert counters["stdout_seen_bytes"] == 300_200, counters assert counters["stdout_retained_bytes"] == 1000 assert counters["stdout_dropped_bytes"] == 299_200 # seen == retained + dropped is the property `head -c` could not give. assert ( counters["stdout_retained_bytes"] + counters["stdout_dropped_bytes"] == counters["stdout_seen_bytes"] ) # The head is kept, and the cut is announced exactly once. assert stdout == "a" * 1000 + sink_drain.truncation_marker(1000) assert stdout.count("...(truncated at") == 1 # A stream that ended under the cap is returned whole, unmarked. assert stderr == "eeeeeee" assert counters["stderr_dropped_bytes"] == 0 def test_the_host_drain_streams_before_the_producer_pauses(): """The criterion the shell drainer failed outright. Fed a writer that emitted 100 bytes, paused 1.2 s and emitted 100 more, `head -c`'s capture file held **0 bytes at 1.8 s** -- its stdout to a file is stdio-buffered, so adopting it would have silently killed the live R output the streaming path exists to provide. Here the first bytes have to arrive while the producer is still asleep. """ directory = sink_drain.SinkDirectory() seen: list[tuple[float, str]] = [] started = time.monotonic() try: capture = directory.open( cap=10_000, on_chunk=lambda text: seen.append((time.monotonic() - started, text)), ) writer = subprocess.Popen( [ sys.executable, "-u", "-c", "import sys, time\n" "out = open(sys.argv[1], 'wb', buffering=0)\n" "out.write(b'a' * 100)\n" "time.sleep(1.2)\n" "out.write(b'b' * 100)\n" "out.close()\n" "open(sys.argv[2], 'wb').close()\n", capture.out_path, capture.err_path, ] ) assert writer.wait(timeout=30) == 0 capture.finish() finally: directory.close() early = [(at, text) for at, text in seen if at < 1.0] assert early, f"nothing reached the host before the producer paused: {seen}" assert early[0][1] == "a" * 100 def test_the_host_drain_does_not_split_a_character_across_two_reads(): """A caller watching a cell live must not see a character become two replacement marks because a multibyte sequence straddled a read boundary. The writer sleeps between the two halves of one character so the reader is guaranteed to see them separately -- without the pause this passes on a fast machine whether or not the decoder is incremental. """ directory = sink_drain.SinkDirectory() seen: list[str] = [] try: capture = directory.open(cap=10_000, on_chunk=seen.append) writer = subprocess.Popen( [ sys.executable, "-u", "-c", "import sys, time\n" "out = open(sys.argv[1], 'wb', buffering=0)\n" # 'é' is two bytes; send them one at a time. "out.write(b'\\xc3')\n" "time.sleep(0.4)\n" "out.write(b'\\xa9!')\n" "out.close()\n" "open(sys.argv[2], 'wb').close()\n", capture.out_path, capture.err_path, ] ) assert writer.wait(timeout=30) == 0 stdout, _ = capture.finish() finally: directory.close() assert "".join(seen) == "é!", seen assert "\ufffd" not in "".join(seen) assert stdout == "é!" @pytest.mark.skipif(_REAL_R is None, reason="no Rscript resolvable on this machine") def test_one_r_expression_cannot_outrun_the_host_cap(tmp_path): """The case the worker-side cap could never reach. R is single threaded, `addTaskCallback` does not fire inside an expression and a connection callback cannot be written in R, so every bound the worker could enforce ran *between* top-level expressions. A cell whose whole body is one expression printing 300 MB wrote all 300 MB to a tempfile -- a tmpfs on much of Linux, so RAM -- and the worker then read 1 MB of it and threw the rest away. Measured on R 4.6.1: 1.2 MB in, 21.2 MB on disk, for output already decided against. Everything below is one expression: the `invisible({...})` braces are what make this test about the gap rather than about the between-expression cap. The counters are the assertion that matters. A capped `stdout` looks identical whether the host read 300 MB and declined 299 of them or the producer quietly dropped them before they were ever written -- and that second reading is not hypothetical, it is exactly what R's fifo() does at its default `blocking = FALSE` (measured: 1.5 MB of 300 MB arrived, no error, status 0). `stdout_seen_bytes` is what tells the two apart. """ from openai4s.kernel.r_kernel import spawn_r_kernel kernel = spawn_r_kernel(cwd=str(tmp_path), rscript=_REAL_R) try: result = kernel.execute( "invisible({cat(strrep('a', 3e8)); cat(strrep('z', 10L))})" ) finally: kernel.shutdown() assert result.get("error") is None, result.get("error") usage = result["usage"] # Every byte the cell wrote reached the host, and the host kept the cap. assert usage["stdout_seen_bytes"] == 300_000_010, usage assert usage["stdout_retained_bytes"] == worker.MAX_OUTPUT assert usage["stdout_dropped_bytes"] == 300_000_010 - worker.MAX_OUTPUT # ...and the answer is still right: head retained, cut announced once. stdout = result["stdout"] assert len(stdout) == worker.MAX_OUTPUT + len( sink_drain.truncation_marker(worker.MAX_OUTPUT) ) assert stdout.count("...(truncated at") == 1 kept = stdout.split("\n...(truncated at")[0] assert set(kept) == {"a"}, sorted(set(kept))[:5] @pytest.mark.skipif(_REAL_R is None, reason="no Rscript resolvable on this machine") def test_the_r_message_stream_is_bounded_by_the_same_drain(tmp_path): """The other stream, asserted separately rather than assumed. `message()` has its own sink and its own fifo. The two are one mechanism called with different arguments, and "wired to the wrong connection" is a defect that looks exactly like "not wired at all" from the stdout side. """ from openai4s.kernel.r_kernel import spawn_r_kernel kernel = spawn_r_kernel(cwd=str(tmp_path), rscript=_REAL_R) try: result = kernel.execute( "invisible({message(strrep('m', 2e7)); message('tail')})" ) finally: kernel.shutdown() assert result.get("error") is None, result.get("error") usage = result["usage"] # 20 MB + the newline message() appends, twice, plus "tail". assert usage["stderr_seen_bytes"] == 20_000_000 + 1 + 4 + 1, usage assert usage["stderr_retained_bytes"] == worker.MAX_OUTPUT assert result["stderr"].count("...(truncated at") == 1 assert set(result["stderr"].split("\n...(truncated at")[0]) == {"m"} # The stdout drain stayed at zero: this cell printed nothing. assert usage["stdout_seen_bytes"] == 0 @pytest.mark.skipif(_REAL_R is None, reason="no Rscript resolvable on this machine") def test_a_finished_r_cell_does_not_wait_out_the_drain_grace(tmp_path): """Every cell pays the drain's exit, so the drain has to end on EOF. The reader cannot tell "no writer yet" from "writer closed" -- both are a zero-length read -- so it carries a bounded grace period for a worker that never closed at all. If EOF detection regressed, nothing would break and nothing would be lost: every R cell in the product would simply take two seconds longer, which no assertion about output would ever notice. Timed on the second cell so the measurement excludes spawning R, and the margin is an order of magnitude: a trivial cell runs in milliseconds and the grace it must not wait for is two seconds. """ from openai4s.kernel.r_kernel import spawn_r_kernel kernel = spawn_r_kernel(cwd=str(tmp_path), rscript=_REAL_R) try: kernel.execute("1 + 1") # warm: spawn and first-cell cost excluded started = time.monotonic() result = kernel.execute("cat('quick\n')") elapsed = time.monotonic() - started finally: kernel.shutdown() assert result["stdout"] == "quick\n" assert elapsed < sink_drain._FINISH_GRACE_SECONDS, ( f"a trivial cell took {elapsed:.2f}s against a " f"{sink_drain._FINISH_GRACE_SECONDS}s grace — the drain is timing out " "rather than ending on EOF" ) # --- the truncation flag, stated rather than derived ------------------------ @pytest.mark.skipif(_REAL_R is None, reason="no Rscript resolvable on this machine") @pytest.mark.parametrize( "size, truncated", [ (0, False), (worker.MAX_OUTPUT - 1, False), (worker.MAX_OUTPUT, False), (worker.MAX_OUTPUT + 1, True), ], ) def test_the_stdout_truncation_flag_is_exact_at_the_cap(tmp_path, size, truncated): """Exactly at the cap is not truncated; one byte past it is. The boundary is the whole value of a boolean here. `dropped > 0` and "the marker is in the text" agree in the easy cases and disagree at the edges -- output that ends exactly at the cap was never cut, and a cell whose own output happens to contain the marker's wording is not evidence of anything. """ from openai4s.kernel.r_kernel import spawn_r_kernel kernel = spawn_r_kernel(cwd=str(tmp_path), rscript=_REAL_R) try: code = "invisible(NULL)" if size == 0 else f"cat(strrep('a', {size}L))" result = kernel.execute(code) finally: kernel.shutdown() assert result.get("error") is None, result.get("error") usage = result["usage"] assert usage["stdout_truncated"] is truncated, usage assert usage["stderr_truncated"] is False, usage assert (usage["stdout_dropped_bytes"] > 0) is truncated @pytest.mark.skipif(_REAL_R is None, reason="no Rscript resolvable on this machine") @pytest.mark.parametrize( "size, truncated", [(worker.MAX_OUTPUT, False), (worker.MAX_OUTPUT + 1, True)], ) def test_the_stderr_truncation_flag_is_exact_at_the_cap(tmp_path, size, truncated): """The other stream, asserted separately: the two are one mechanism called with different arguments, and "wired to the wrong connection" looks exactly like "not wired at all" from the stdout side.""" from openai4s.kernel.r_kernel import spawn_r_kernel kernel = spawn_r_kernel(cwd=str(tmp_path), rscript=_REAL_R) try: # `message()` appends a newline, so ask for one byte less than the size # under test and let R supply the last one. result = kernel.execute(f"message(strrep('m', {size - 1}L))") finally: kernel.shutdown() assert result.get("error") is None, result.get("error") usage = result["usage"] assert usage["stderr_truncated"] is truncated, usage assert usage["stdout_truncated"] is False, usage @pytest.mark.skipif(_REAL_R is None, reason="no Rscript resolvable on this machine") def test_a_multibyte_character_cut_by_the_cap_is_still_reported_truncated(tmp_path): """The cap counts bytes and the text is characters, so a cut can land in the middle of one. The flag must not depend on the repaired text: the replacement character is what a split looks like *after* decoding, and reading truncation off the string would make that a guess.""" from openai4s.kernel.r_kernel import spawn_r_kernel kernel = spawn_r_kernel(cwd=str(tmp_path), rscript=_REAL_R) try: # Three-byte characters, so the cap cannot fall on a boundary. count = (worker.MAX_OUTPUT // 3) + 10 result = kernel.execute(f"cat(strrep('\\u4e2d', {count}L))") finally: kernel.shutdown() assert result.get("error") is None, result.get("error") usage = result["usage"] assert usage["stdout_truncated"] is True, usage assert usage["stdout_dropped_bytes"] > 0 assert result["stdout"].count("...(truncated at") == 1 @pytest.mark.skipif(_REAL_R is None, reason="no Rscript resolvable on this machine") def test_an_interrupted_cell_reports_what_the_host_had_drained(tmp_path): """A real interrupt against the real fifo, not the parser or a helper. SIGINT during a flooding expression is the case where the writer stops mid-stream: R unwinds its sinks and closes, the host's readers see EOF, and what they had already taken is what the cell reports. Driving the frame parser instead would prove the protocol and say nothing about whether the drain survives its writer being killed. The interrupt is armed by the first drained chunk, not by a timer: a conda Rscript can take seconds to start, and a fixed delay that expires first lands the SIGINT before the cell has streamed a byte — outside r_worker.R's handler it halts the interpreter, inside it it interrupts a cell whose `stdout_seen_bytes` is legitimately 0. Both are startup races, not the mid-stream case. `on_chunk` fires on the drain thread after `seen` was bumped, so the SIGINT is sent only when bytes have provably crossed the fifo while the writer still has ~4 GB left to produce. """ from openai4s.kernel.r_kernel import spawn_r_kernel # This test once skipped when its own process had SIGINT set to SIG_IGN (a # backgrounded `pytest &`): SIG_IGN survived exec into R, R honoured it, # and no SIGINT could reach the worker. The spawn boundary in # kernel/transport.py now resets the child's disposition, so the scenario # runs instead of skipping — test_r_kernel.py pins that reset explicitly. kernel = spawn_r_kernel(cwd=str(tmp_path), rscript=_REAL_R) try: fired = threading.Event() def interrupt_once_streaming(_text: str) -> None: # Exactly one SIGINT: the drain can keep delivering chunks after # the first, and a second signal could land between cells. if not fired.is_set(): fired.set() kernel.interrupt() result = kernel.execute( "invisible(lapply(1:4000, function(i) cat(strrep('x', 1e6))))", on_chunk=interrupt_once_streaming, ) assert fired.is_set(), "the cell finished without streaming a chunk" assert result.get("interrupted") is True, result usage = result["usage"] # The host drained real bytes before the interrupt landed, and the # counters describe them rather than being reset by the interrupt. assert usage["stdout_seen_bytes"] > 0, usage assert usage["stdout_retained_bytes"] <= worker.MAX_OUTPUT # ...and the kernel is still usable, which is what distinguishes an # interrupt from a crash. after = kernel.execute("cat('still here\n')") assert after["stdout"] == "still here\n" assert after["usage"]["stdout_truncated"] is False finally: kernel.shutdown() #: R, run inside a cell: bind ``worker_private`` to the worker's private #: environment, found through the worker functions on the cell's own call stack. #: The worker keeps nothing in ``.GlobalEnv`` (a cell clearing it must not #: delete the kernel), so this is how a test reaches a helper to replace it. _R_FIND_WORKER_ENV = """ worker_private <- (function() { for (k in rev(seq_len(sys.nframe()))) { env <- environment(sys.function(k)) if (is.environment(env) && exists('.oai4s_handle_line', envir = env, inherits = FALSE)) return(env) } stop('no worker environment on the call stack') })() """ @pytest.mark.skipif(_REAL_R is None, reason="no Rscript resolvable on this machine") @pytest.mark.parametrize( ("raise_condition", "expected_error", "interrupted"), [ ( "signalCondition(structure(list(message = 'forced'), " "class = c('interrupt', 'condition')))", "Interrupted", True, ), ( "stop('forced')", "openai4s r_worker internal error: forced", False, ), ], ids=("interrupted", "internal-error"), ) def test_r_top_level_fallback_preserves_host_capture( tmp_path, raise_condition, expected_error, interrupted ): """Both outer fallbacks return output already drained by the host. A timed SIGINT can land inside ``.oai4s_run`` and exercise its normal response. Replace the dispatcher for one frame instead: it writes known bytes to the real host FIFOs, restores itself, then raises outside the cell handler so the top-level interrupt/error fallback must respond. The dispatcher is not a ``.GlobalEnv`` binding -- the worker keeps its state in a private environment so a cell clearing the user namespace cannot delete it -- so the patch reaches that environment through the worker functions on this cell's own call stack. """ from openai4s.kernel.r_kernel import spawn_r_kernel kernel = spawn_r_kernel(cwd=str(tmp_path), rscript=_REAL_R) try: setup = f""" {_R_FIND_WORKER_ENV} .oai4s_saved_handle_line <- get('.oai4s_handle_line', envir = worker_private) assign('.oai4s_handle_line', function(line) {{ frame <- jsonlite::fromJSON(line, simplifyVector = TRUE) if (is.null(frame$sink_out)) {{ # Only an execute frame carries sinks. Hand anything else back rather than # crash on fifo(NULL): the self-restore below runs on the FIRST call, so an # assertion failing before it would otherwise leave this patch installed to # swallow teardown's shutdown frame, and kernel.shutdown() would spend its # whole 5 s budget before SIGKILLing a worker that never saw the request. return(.oai4s_saved_handle_line(line)) }} assign('.oai4s_handle_line', .oai4s_saved_handle_line, envir = worker_private) out_con <- fifo(as.character(frame$sink_out), open = 'wb', blocking = TRUE) err_con <- fifo(as.character(frame$sink_err), open = 'wb', blocking = TRUE) writeBin(charToRaw('fallback stdout\\n'), out_con) writeBin(charToRaw('fallback stderr\\n'), err_con) flush(out_con); flush(err_con) close(out_con); close(err_con) {raise_condition} }}, envir = worker_private) """ primed = kernel.execute(setup) assert primed["error"] is None, primed result = kernel.execute("ignored") usage = result["usage"] assert result["interrupted"] is interrupted assert result["error"] == expected_error assert result["sink_capture"] is True assert result["stdout"] == "fallback stdout\n" assert result["stderr"] == "fallback stderr\n" # The byte counters are the host's and are real; the worker's own three # are not measured on this path. `peak_rss_kb` says so with null rather # than 0 -- a fabricated zero beside real numbers reads as measured, and # lands in execution_log.peak_rss_kb as one. wall_s/cpu_s cannot say it # yet: .oai4s_num renders NULL as 0, so they remain the one place this # frame still asserts a measurement it did not take. assert usage == { "wall_s": 0.0, "cpu_s": 0.0, "peak_rss_kb": None, **sink_drain.channel_counters(seen=16, retained=16, prefix="stdout_"), **sink_drain.channel_counters(seen=16, retained=16, prefix="stderr_"), } after = kernel.execute("cat('still here\\n')") assert after["stdout"] == "still here\n" assert after["usage"]["stdout_truncated"] is False finally: kernel.shutdown() @pytest.mark.skipif(_REAL_R is None, reason="no Rscript resolvable on this machine") @pytest.mark.parametrize("failing_helper", (".oai4s_unwind_sinks", ".oai4s_respond")) def test_r_top_level_fallback_cannot_halt_rscript_or_hang_the_host( tmp_path, failing_helper ): """The fallback's own helpers failing is a reported outcome, never a halt. The fallback runs at the read loop's top level, where nothing else catches a condition. Its sink unwind and its response were called bare, so when either raised -- as both did once a cell had deleted them -- Rscript halted and the session's state went with it. - The unwind failing costs nothing: the frame is still answered and the same process serves the next cell. - The response itself failing leaves nothing that can answer the frame. Swallowing that and reading on would leave the manager blocked on an id that never arrives, which is a hang; the worker exits instead, and the host reports a dead worker promptly. """ from openai4s.kernel.r_kernel import spawn_r_kernel kernel = spawn_r_kernel(cwd=str(tmp_path), rscript=_REAL_R) try: setup = f""" {_R_FIND_WORKER_ENV} saved_handle_line <- get('.oai4s_handle_line', envir = worker_private) saved_helper <- get('{failing_helper}', envir = worker_private) assign('.oai4s_handle_line', function(line) {{ frame <- jsonlite::fromJSON(line, simplifyVector = TRUE) if (is.null(frame$sink_out)) return(saved_handle_line(line)) assign('.oai4s_handle_line', saved_handle_line, envir = worker_private) out_con <- fifo(as.character(frame$sink_out), open = 'wb', blocking = TRUE) err_con <- fifo(as.character(frame$sink_err), open = 'wb', blocking = TRUE) close(out_con); close(err_con) assign('{failing_helper}', function(...) {{ assign('{failing_helper}', saved_helper, envir = worker_private) stop('helper failed') }}, envir = worker_private) stop('forced') }}, envir = worker_private) """ primed = kernel.execute(setup) assert primed["error"] is None, primed pid = kernel.pid outcome: dict = {} def run() -> None: try: outcome["result"] = kernel.execute("ignored") except RuntimeError as error: outcome["error"] = str(error) runner = threading.Thread(target=run, daemon=True) runner.start() runner.join(60) if runner.is_alive(): os.kill(pid, 9) runner.join(10) pytest.fail("the worker neither answered the frame nor exited") if failing_helper == ".oai4s_unwind_sinks": assert "error" not in outcome, outcome assert outcome["result"]["error"] == ( "openai4s r_worker internal error: forced" ) assert kernel.is_alive() and kernel.pid == pid after = kernel.execute("cat('still here\\n')") assert after["stdout"] == "still here\n" else: assert "exited unexpectedly" in outcome.get("error", ""), outcome assert not kernel.is_alive() finally: kernel.shutdown() @pytest.mark.skipif(_REAL_R is None, reason="no Rscript resolvable on this machine") def test_an_interrupt_escaping_oai4s_run_still_reports_the_host_capture(tmp_path): """The fallback shape production actually produces, driven end to end. The test above replaces the dispatcher, so it opens the fifos itself and closes them before raising -- the host's readers then end on EOF. The real escape cannot do that: ``out_con``/``msg_con`` are locals of the ``.oai4s_run`` frame the condition unwound, and ``.oai4s_unwind_sinks()`` pops sink levels without closing connections. Reach it the way a user does -- ``.oai4s_run`` autoprints a visible value at a ``tryCatch`` that handles ``error`` but not ``interrupt`` -- so the sinks are still live and the cell really wrote through them. Deterministic: the condition is signalled by the print method, not by a timed SIGINT. """ from openai4s.kernel.r_kernel import spawn_r_kernel kernel = spawn_r_kernel(cwd=str(tmp_path), rscript=_REAL_R) try: primed = kernel.execute( "print.oai4sprobe <- function(x, ...) {\n" " cat('written through the live sink\\n')\n" " signalCondition(structure(list(message = 'ctrl-c')," " class = c('interrupt', 'condition')))\n" "}\n" "probe <- function() structure(list(), class = 'oai4sprobe')\n" ) assert primed["error"] is None, primed result = kernel.execute("probe()") assert result["interrupted"] is True assert result["error"] == "Interrupted" # The point of the fix: what the cell wrote before the escape survives, # and it is the host that has it. assert result["sink_capture"] is True assert result["stdout"] == "written through the live sink\n" assert result["usage"]["stdout_seen_bytes"] == 30 assert result["usage"]["stdout_truncated"] is False # Not a measurement this path took -- absent, not zero. assert result["usage"]["peak_rss_kb"] is None after = kernel.execute("cat('still here\\n')") assert after["stdout"] == "still here\n" finally: kernel.shutdown() # --- the kernel's own stderr pipe ------------------------------------------ def test_the_kernel_stderr_tail_is_bounded_in_bytes_not_lines(): """`deque(maxlen=400)` bounded the number of lines and nothing else. The drain site's own comment names the producers that reach it -- an R `system()`, an uncaptured subprocess writing to inherited fd2 -- and what they emit has no newline discipline. `for line in stream` is `readline()`: it allocates until it finds a newline, so one newline-free blob is allocated in full *before* any cap applies, and a 400-line ceiling never fires for it. Measured: a single 20 MB line left `deque(maxlen=400)` holding 20 MB at one line of four hundred, comfortably inside its limit. """ from openai4s.kernel.manager import _STDERR_TAIL_BYTES, _StderrTail tail = _StderrTail(_STDERR_TAIL_BYTES) blob = b"x" * (20 * 1024 * 1024) for offset in range(0, len(blob), 8192): tail.feed(blob[offset : offset + 8192]) assert tail.retained_bytes == _STDERR_TAIL_BYTES assert tail.seen_bytes == len(blob) assert tail.dropped_bytes == len(blob) - _STDERR_TAIL_BYTES assert tail.truncated is True # The death path joins it; that call site is unchanged. assert len("".join(tail)) == _STDERR_TAIL_BYTES def test_the_stderr_tail_keeps_the_end_and_reports_an_untruncated_read(): """The tail is what explains a death, and a short stream is not truncated.""" from openai4s.kernel.manager import _StderrTail tail = _StderrTail(64) tail.feed(b"a" * 40) assert tail.truncated is False assert tail.dropped_bytes == 0 assert tail.text() == "a" * 40 tail.feed(b"b" * 40) assert tail.truncated is True assert tail.text().endswith("b" * 40) assert tail.retained_bytes == 64 def test_the_drain_reads_the_raw_pipe_so_the_budget_is_in_real_bytes(): """The budget must be counted on bytes that actually arrived. `text=True` is required for the stdout protocol frames and is per-Popen, so the stderr budget has to sit on the `BufferedReader` underneath the decoder. Counting characters off a decoded stream is the unit confusion `mcp_client` was rewritten to remove: four million three-byte characters is twelve megabytes, not four. Behavioural rather than a source-text match -- an earlier draft asserted on the source and tripped over its own variable name. """ import subprocess from openai4s.kernel.manager import _StderrTail # A three-byte character per unit: a character budget would keep three # times what a byte budget does. proc = subprocess.Popen( [sys.executable, "-c", "import sys; sys.stderr.write('\u4e2d' * 200)"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, bufsize=1, ) try: fd = proc.stderr.fileno() tail = _StderrTail(120) while True: chunk = os.read(fd, 8192) if not chunk: break assert isinstance(chunk, bytes), "the drain decoded before counting" tail.feed(chunk) finally: proc.wait(timeout=10) # 200 characters is 600 bytes; the 120-byte budget saw all of them. assert tail.seen_bytes == 600 assert tail.retained_bytes == 120 assert tail.truncated is True def test_the_spawn_wires_the_drain_to_the_raw_pipe(): """That the production drain reads the raw pipe, not just that it could. The test above builds its own reader off `.buffer` and proves the byte budget works there. Reverting `_spawn` to read the decoded `proc.stderr` left it green -- it verifies the mechanism and not the wiring, which is the gap this codebase keeps shipping. Asserted on the compiled code rather than the source text: an earlier draft matched a source string and broke on a variable rename, and spawning a real kernel here hangs the suite. The target moved from `Kernel._spawn` to `PipeTransport` when the kernel grew a second transport (M3b-1). The claim did not move: this is still about the production wiring for a local child, which is where every drain the product actually runs comes from. """ from openai4s.kernel.transport import PipeTransport spawn = PipeTransport._start_stderr_drain assert "fileno" in spawn.__code__.co_names, ( "the drain no longer takes the raw descriptor, so its byte budget is " "counted on decoded text" ) nested = [c for c in spawn.__code__.co_consts if hasattr(c, "co_names")] drain = [c for c in nested if "feed" in c.co_names] assert drain, "the drain does not feed the bounded tail" # `os.read`, not a buffered read. A daemon thread parked inside a buffered # read holds that buffer's lock when the interpreter finalises: the whole # suite passed and then aborted with `_enter_buffered_busy: could not # acquire lock ... at interpreter shutdown`. A clean exit turned into # SIGABRT by the drain alone, and only the full-suite gate saw it. assert any("os" in c.co_names and "read" in c.co_names for c in drain), ( "the drain uses a buffered read; a daemon thread must not hold a " "BufferedReader lock at interpreter shutdown" )