"""Kernel tests: persistent namespace, print capture, error attribution, usage accounting, and host_call RPC round-trip (dispatcher stubbed).""" import ntpath import os import signal import sys import threading import time from concurrent.futures import ThreadPoolExecutor from pathlib import Path import pytest from openai4s.config import Config, LLMConfig from openai4s.host_dispatch import build_dispatcher from openai4s.kernel import Kernel, KernelBusyError from openai4s.kernel import manager as manager_mod from openai4s.kernel import worker as worker_mod from openai4s.kernel.environment import build_kernel_environment def _echo_dispatcher(method, args): if method == "ping": return "pong" if method == "add": return sum(args[0]["nums"]) raise ValueError(f"unknown method {method}") def _authorized_bash_dispatcher(tmp_path): """Real Host policy/authorization with an explicit test-only allow rule.""" cfg = Config( data_dir=tmp_path / ".data", llm=LLMConfig(provider="deepseek", api_key="test-only"), ) dispatcher = build_dispatcher(cfg, workspace=tmp_path) frame_id = dispatcher.store.new_frame(kind="turn") dispatcher.frame_id = frame_id dispatcher.store.set_permission_rule( scope="conversation", scope_id=frame_id, tool="bash", pattern="*", decision="allow", ) return dispatcher def test_print_capture(): with Kernel(dispatcher=_echo_dispatcher) as k: r = k.execute("print('hello')") assert r["stdout"] == "hello\n" assert r["error"] is None def test_persistent_namespace(): with Kernel(dispatcher=_echo_dispatcher) as k: k.execute("x = 41") r = k.execute("print(x + 1)") assert r["stdout"].strip() == "42" #: A worker's one-time cold start -- the sandbox self-test, the Host facade #: and provenance install on its first Cell -- is not what the thread-affinity #: test below measures, and on a loaded macOS runner it alone has taken 17-48 #: seconds. It gets its own generous budget. _COLD_START_BUDGET_S = 120 #: The property under test: a warm kernel answers its creating request thread. _WARM_EXECUTE_BUDGET_S = 15 def test_kernel_survives_the_request_thread_that_created_it(tmp_path): kernel = None try: with ThreadPoolExecutor(max_workers=1) as requests: kernel = requests.submit(Kernel, cwd=str(tmp_path)).result( timeout=_COLD_START_BUDGET_S ) warm = requests.submit(kernel.execute, "pass").result( timeout=_COLD_START_BUDGET_S ) assert warm["error"] is None first = requests.submit(kernel.execute, "marker = 41").result( timeout=_WARM_EXECUTE_BUDGET_S ) assert first["error"] is None # The request thread is now gone, as after a Web REPL/Agent turn. The # next request must see the same worker and its original namespace. time.sleep(0.1) result = kernel.execute("print(marker + 1)") assert result["error"] is None assert result["stdout"].strip() == "42" finally: if kernel is not None: kernel.shutdown() def test_files_read_are_runtime_observations_not_source_guesses(tmp_path): (tmp_path / "executed.txt").write_text("observed", encoding="utf-8") (tmp_path / "dead.txt").write_text("not observed", encoding="utf-8") with Kernel(dispatcher=_echo_dispatcher, cwd=str(tmp_path)) as kernel: result = kernel.execute( "from pathlib import Path\n" "if False:\n" " Path('dead.txt').read_text()\n" "print(Path('executed.txt').read_text())\n" "Path('write-only.txt').write_text('output')" ) assert result["error"] is None assert result["stdout"].strip() == "observed" assert result["files_read"] == ["executed.txt"] assert (tmp_path / "write-only.txt").read_text(encoding="utf-8") == "output" def test_files_read_normalize_workspace_paths_and_exclude_external_reads(tmp_path): nested = tmp_path / "nested" nested.mkdir() (nested / "input.bin").write_bytes(b"data") with Kernel(dispatcher=_echo_dispatcher, cwd=str(tmp_path)) as kernel: result = kernel.execute( "import os\n" "open(os.path.join('nested', '..', 'nested', 'input.bin'), 'rb').read()\n" "open('/dev/null', 'rb').read()\n" "fd = os.open('created.bin', os.O_CREAT | os.O_WRONLY, 0o600)\n" "os.close(fd)" ) assert result["error"] is None assert result["files_read"] == ["nested/input.bin"] def test_file_read_path_normalization_uses_windows_rules_without_posix_imports(): def relative(raw_path, *, cwd=r"C:\Workspace\analysis"): return worker_mod._workspace_relative_read_path( raw_path, root=r"C:\Workspace", cwd=cwd, isabs=ntpath.isabs, normpath=ntpath.normpath, relpath=ntpath.relpath, separator="\\", ) assert relative(r"..\Data\input.csv") == "Data/input.csv" assert relative(r"c:\workspace\Data\input.csv") == "Data/input.csv" assert relative(r"link\input.csv") == "analysis/link/input.csv" assert relative(r"C:\Other\secret.csv") is None assert relative(r"D:\Workspace\input.csv") is None assert relative(r"C:\Workspace", cwd=r"C:\Workspace") is None def test_variable_inspector_reads_only_safe_builtins_without_repr_hooks(): with Kernel(dispatcher=_echo_dispatcher) as kernel: setup = kernel.execute( "events = []\n" "class Meta(type):\n" " def __getattribute__(cls, name):\n" " if name == '__name__': events.append('metaclass-name')\n" " return super().__getattribute__(name)\n" "class Hostile(metaclass=Meta):\n" " def __repr__(self): events.append('repr'); raise RuntimeError('repr')\n" " def __len__(self): events.append('len'); raise RuntimeError('len')\n" " def __sizeof__(self): events.append('sizeof'); raise RuntimeError('sizeof')\n" "class TrapList(list):\n" " def __repr__(self): events.append('list-repr'); raise RuntimeError('repr')\n" " def __iter__(self): events.append('list-iter'); raise RuntimeError('iter')\n" " def __len__(self): events.append('list-len'); raise RuntimeError('len')\n" "score = 0.93\n" "title = 'protein'\n" "samples = [1, 2, 3]\n" "mixed = [1, Hostile()]\n" "hostile = Hostile()\n" "trap = TrapList([1, 2])" ) assert setup["error"] is None response = kernel.inspect_variables() variables = {item["name"]: item for item in response["variables"]} assert variables["score"]["preview"] == 0.93 assert variables["title"]["preview"] == "protein" assert variables["samples"]["length"] == 3 assert len(variables["samples"]["fingerprint"]) == 64 assert variables["mixed"] == { "name": "mixed", "type": "list", "kind": "container", "length": 2, } assert variables["hostile"] == {"name": "hostile", "type": "Hostile"} assert variables["trap"] == {"name": "trap", "type": "TrapList"} assert "host" not in variables and "openai4s" not in variables assert kernel.execute("print(events)")["stdout"].strip() == "[]" def test_variable_inspector_hides_the_skill_bootstrap_it_did_not_author(tmp_path): """The Variable Inspector lists the user's bindings, not the session's glue. Every Web session runs the skill import gate as a system cell before the first user cell, and that source binds its helpers into the same namespace user code runs in -- as it must: the frozen-sidecar replay reads its policy back through ``globals()``. The inspector hid only dunders, ``host`` and ``openai4s``, so a session holding one variable listed 33, thirty-two of them ``_o4s_*`` imports, sets and classes the user never wrote. Driven through the same bootstrap call the gateway makes, so the names under test are the ones a real session injects rather than a list copied from the generator. """ from openai4s.server.recovery_runtime import bootstrap_python_generation from openai4s.skills_loader import SkillLoader with Kernel(dispatcher=_echo_dispatcher, cwd=str(tmp_path)) as kernel: metadata = bootstrap_python_generation( kernel, tmp_path, SkillLoader().bootstrap_code() ) assert metadata["status"] == "active", metadata assert kernel.execute("pp = 7")["error"] is None names = [item["name"] for item in kernel.inspect_variables()["variables"]] assert names == ["pp"] def test_variable_inspector_fails_busy_without_competing_frame_reader(): with Kernel(dispatcher=_echo_dispatcher) as kernel: started = threading.Event() result = {} def run(): result["cell"] = kernel.execute( "print('started')\nimport time\ntime.sleep(30)", on_chunk=lambda _text: started.set(), ) thread = threading.Thread(target=run, daemon=True) thread.start() # A completely cold optional-science install may build Matplotlib's # font cache before the first user byte is emitted. The cache now has a # stable writable location, but this concurrency assertion should not # mistake that one-time initialization for a protocol deadlock. assert started.wait(60) with pytest.raises(KernelBusyError, match="busy"): kernel.inspect_variables() kernel.interrupt() thread.join(timeout=15) assert not thread.is_alive() assert result["cell"]["interrupted"] is True # The failed busy read consumed no frame; the next request is aligned. assert isinstance(kernel.inspect_variables()["variables"], list) def test_variable_inspector_limit_validation_is_local(): with Kernel(dispatcher=_echo_dispatcher) as kernel: with pytest.raises(ValueError, match="between 1 and 500"): kernel.inspect_variables(limit=0) with pytest.raises(TypeError, match="integer"): kernel.inspect_variables(limit=True) assert kernel.execute("print('aligned')")["stdout"].strip() == "aligned" class _InitializationFailureTransport: """A transport peer that fails before a kernel candidate can be published.""" def __init__(self, failure: str) -> None: self.failure = failure self.process = None self.stderr_tail = None self.closed = False self.killed = False self._released = threading.Event() def write_line(self, _line: str) -> None: if self.failure == "write": raise OSError("initialization write failed") def read_line(self) -> str: if self.failure == "read": raise OSError("initialization read failed") self._released.wait(timeout=5) return "" def alive(self) -> bool: return not self.closed def interrupt(self) -> bool: return False def kill(self) -> None: self.killed = True self._released.set() def close(self, *, graceful: bool = True) -> None: self.closed = True self._released.set() @pytest.mark.parametrize("failure", ["write", "read"]) def test_attestation_initialization_errors_close_the_transport(failure): transport = _InitializationFailureTransport(failure) with pytest.raises(RuntimeError, match="attestation initialization") as caught: Kernel(transport_factory=lambda: transport) assert isinstance(caught.value.__cause__, OSError) assert transport.closed is True def test_attestation_initialization_deadline_kills_the_blocked_transport( monkeypatch, ): transport = _InitializationFailureTransport("block") monkeypatch.setattr(manager_mod, "_SKILL_SIDECAR_INITIALIZATION_TIMEOUT_S", 0.01) with pytest.raises(RuntimeError, match="deadline"): Kernel(transport_factory=lambda: transport) assert transport.killed is True assert transport.closed is True def test_kernel_child_environment_is_rebuilt_from_strict_allowlist(tmp_path): source = { "PATH": "/host/bin", "HOME": "/home/scientist", "LANG": "en_US.UTF-8", "TMPDIR": "/safe/tmp", "MPLBACKEND": "Agg", "VIRTUAL_ENV": "/host/venv", "PYTHONPATH": "/host/injected-pythonpath", "CONDA_PREFIX": "/host/wrong-conda", "OPENAI4S_PROVENANCE_OFF": "1", "OPENAI4S_SAFETY_AUDIT_HOOK": "0", "OPENAI4S_KERNEL_MODE": "host-override", "OPENAI4S_KERNEL_GENERATION": "host-forged-generation", "OPENAI4S_WORKSPACE": "/host/wrong-workspace", "OPENAI4S_LLM_API_KEY": "llm-secret", "OPENAI4S_ARK_API_KEY": "ark-secret", "OPENAI_API_KEY": "openai-secret", "ANTHROPIC_API_KEY": "anthropic-secret", "HF_TOKEN": "hf-secret", "AWS_SECRET_ACCESS_KEY": "aws-secret", "SYNTHETIC_OAUTH_CREDENTIAL": "oauth-secret", "DATABASE_PASSWORD": "db-secret", "SSH_AUTH_SOCK": "/tmp/agent.sock", "HTTPS_PROXY": "https://user:password@proxy.invalid", "LD_PRELOAD": "/tmp/evil.so", "LD_LIBRARY_PATH": "/tmp/evil-libs", "DYLD_INSERT_LIBRARIES": "/tmp/evil.dylib", "DYLD_LIBRARY_PATH": "/tmp/evil-dylibs", "BASH_ENV": "/tmp/evil-bashrc", "NODE_OPTIONS": "--require=/tmp/evil.js", } repo_root = tmp_path / "trusted-repo" env_root = tmp_path / "conda" / "science" env = build_kernel_environment( source=source, mode="python", cwd=str(tmp_path), env_root=str(env_root), env_name="science", kernel_generation="kernel:test-generation", repo_root=str(repo_root), ) assert env["PATH"].split(os.pathsep)[0] == str(env_root / "bin") assert env["HOME"] == "/home/scientist" assert env["LANG"] == "en_US.UTF-8" assert env["TMPDIR"] == "/safe/tmp" assert env["MPLBACKEND"] == "Agg" assert env["MPLCONFIGDIR"] == "/safe/tmp/openai4s-matplotlib" assert env["OPENAI4S_PROVENANCE_OFF"] == "1" assert env["OPENAI4S_SAFETY_AUDIT_HOOK"] == "0" assert env["OPENAI4S_KERNEL_MODE"] == "python" assert env["OPENAI4S_KERNEL_GENERATION"] == "kernel:test-generation" assert env["OPENAI4S_WORKSPACE"] == str(tmp_path.resolve()) assert env["PWD"] == str(tmp_path.resolve()) assert env["PYTHONPATH"] == str(repo_root.resolve()) assert env["CONDA_PREFIX"] == str(env_root) assert env["CONDA_DEFAULT_ENV"] == "science" assert env["CONDA_SHLVL"] == "1" assert "VIRTUAL_ENV" not in env # selected conda runtime wins forbidden = { "OPENAI4S_LLM_API_KEY", "OPENAI4S_ARK_API_KEY", "OPENAI_API_KEY", "ANTHROPIC_API_KEY", "HF_TOKEN", "AWS_SECRET_ACCESS_KEY", "SYNTHETIC_OAUTH_CREDENTIAL", "DATABASE_PASSWORD", "SSH_AUTH_SOCK", "HTTPS_PROXY", "LD_PRELOAD", "LD_LIBRARY_PATH", "DYLD_INSERT_LIBRARIES", "DYLD_LIBRARY_PATH", "BASH_ENV", "NODE_OPTIONS", } assert forbidden.isdisjoint(env) assert "/host/injected-pythonpath" not in env["PYTHONPATH"] assert source["OPENAI4S_LLM_API_KEY"] == "llm-secret" # source not mutated def test_the_kernel_interpreter_bin_dir_leads_path_when_no_env_is_selected(tmp_path): """A shell a Cell starts must resolve `python` to the Cell's own interpreter. Only a selected conda prefix used to reach PATH, so a daemon or CLI launched by absolute path from a non-activated venv gave `host.bash("python -m pytest -q")` whatever `python` the host PATH had -- nothing at all on macOS (exit 127).""" source = {"PATH": "/usr/bin:/bin", "HOME": "/home/scientist"} interpreter = str(tmp_path / "venv" / "bin" / "python") env = build_kernel_environment( source=source, cwd=str(tmp_path), interpreter=interpreter ) assert env["PATH"].split(os.pathsep) == [ str(tmp_path / "venv" / "bin"), "/usr/bin", "/bin", ] # Unresolved: a venv interpreter is often a symlink to a base Python, and # the base Python's directory has none of the venv's site-packages. linked = tmp_path / "linked-venv" / "bin" linked.mkdir(parents=True) (linked / "python").symlink_to(sys.executable) env = build_kernel_environment( source=source, cwd=str(tmp_path), interpreter=str(linked / "python") ) assert env["PATH"].split(os.pathsep)[0] == str(linked) # A selected conda prefix still wins, and a bare or relative interpreter # never puts the workspace ('.') on PATH. conda = tmp_path / "conda" / "science" env = build_kernel_environment( source=source, cwd=str(tmp_path), env_root=str(conda), interpreter=interpreter, ) assert env["PATH"].split(os.pathsep) == [str(conda / "bin"), "/usr/bin", "/bin"] for bare in ("python", "./python", ""): env = build_kernel_environment( source=source, cwd=str(tmp_path), interpreter=bare ) assert env["PATH"] == "/usr/bin:/bin" assert build_kernel_environment(source=source, cwd=str(tmp_path))["PATH"] == ( "/usr/bin:/bin" ) def test_an_r_worker_path_is_not_given_the_daemon_python(monkeypatch, tmp_path): """The R kernel reuses this manager with its own argv; `self.python` is still the daemon's interpreter there, which R's shell must not inherit.""" # Sparse, as in a non-activated launch: `uv run` would otherwise already # have put the venv's bin dir first and this would prove nothing. monkeypatch.setenv("PATH", "/usr/bin:/bin") def child_env(argv): kernel = manager_mod.Kernel.__new__(manager_mod.Kernel) kernel.mode = "repl" kernel.cwd = str(tmp_path) kernel.python = sys.executable kernel.env_root = None kernel.env_name = None kernel.argv = argv kernel.authorization_generation = "kernel:test" return kernel._child_env() python_dir = os.path.dirname(sys.executable) assert child_env(None)["PATH"].split(os.pathsep)[0] == python_dir r_path = child_env(["sh", "-c", "exec Rscript r_worker.R"])["PATH"] assert r_path == "/usr/bin:/bin" def test_host_bash_python_is_the_kernel_interpreter_in_a_non_activated_launch( monkeypatch, tmp_path ): """The command the explicit code modes teach, run the way they teach it, from a launch with no venv activated and no `python` on the host PATH.""" monkeypatch.setenv("PATH", "/usr/bin:/bin") monkeypatch.delenv("VIRTUAL_ENV", raising=False) code = """ import sys version = host.bash('python -m pytest --version') print('pytest-exit', version['exit_code']) probe = host.bash('python -c "import sys; print(sys.prefix)"') print('same-prefix', probe['stdout'].strip() == sys.prefix) """ with Kernel( dispatcher=_authorized_bash_dispatcher(tmp_path), cwd=str(tmp_path) ) as kernel: result = kernel.execute(code) assert result["error"] is None, result assert "pytest-exit 0" in result["stdout"], result assert "same-prefix True" in result["stdout"], result def test_python_kernel_and_its_subprocesses_cannot_inherit_host_api_key( monkeypatch, tmp_path ): marker = "synthetic-host-llm-secret-never-in-kernel" monkeypatch.setenv("OPENAI4S_LLM_API_KEY", marker) monkeypatch.setenv("OPENAI4S_ARK_API_KEY", marker) monkeypatch.setenv("OPENAI_API_KEY", marker) monkeypatch.setenv("HF_TOKEN", marker) monkeypatch.setenv("LD_PRELOAD", "/tmp/openai4s-never-load.so") monkeypatch.setenv("DYLD_INSERT_LIBRARIES", "/tmp/openai4s-never-load.dylib") monkeypatch.setenv("OPENAI4S_PROVENANCE_OFF", "1") code = """ import os, shlex, subprocess, sys probe = "import os; print(os.environ.get('OPENAI4S_LLM_API_KEY', ''))" print(os.environ.get('OPENAI4S_LLM_API_KEY', '')) print(subprocess.check_output([sys.executable, '-c', probe], text=True).strip()) cmd = shlex.quote(sys.executable) + ' -c ' + shlex.quote(probe) print(host.bash(cmd)['stdout'].strip()) print(os.environ.get('OPENAI4S_PROVENANCE_OFF', '')) """ with Kernel( dispatcher=_authorized_bash_dispatcher(tmp_path), cwd=str(tmp_path) ) as kernel: result = kernel.execute(code) assert result["error"] is None assert result["stdout"].splitlines() == ["", "", "", "1"] assert marker not in result["stdout"] def test_expr_echo(): with Kernel(dispatcher=_echo_dispatcher) as k: r = k.execute("21 * 2") assert r["stdout"].strip() == "42" def test_error_lineno(): with Kernel(dispatcher=_echo_dispatcher) as k: r = k.execute("a = 1\nb = 2\nraise ValueError('boom')") assert r["error"] is not None assert "ValueError" in r["error"] assert r["trace"]["error_lineno"] == 3 def test_usage_accounting(): with Kernel(dispatcher=_echo_dispatcher) as k: r = k.execute("sum(range(1000))") u = r["usage"] assert set(u) == {"wall_s", "cpu_s", "peak_rss_kb"} assert u["wall_s"] >= 0 and u["peak_rss_kb"] > 0 def test_host_call_roundtrip(): with Kernel(dispatcher=_echo_dispatcher) as k: r = k.execute("reply = host._call('ping', [])\n" "print(reply)") assert r["stdout"].strip() == "pong" def test_host_call_with_args(): with Kernel(dispatcher=_echo_dispatcher) as k: r = k.execute("print(host._call('add', [{'nums': [1, 2, 3, 4]}]))") assert r["stdout"].strip() == "10" def test_host_call_error_propagates(): with Kernel(dispatcher=_echo_dispatcher) as k: r = k.execute( "try:\n" " host._call('nope', [])\n" "except RuntimeError as e:\n" " print('caught:', 'unknown method' in str(e))" ) assert r["stdout"].strip() == "caught: True" # --- frame-protocol contract tests (PR 10) --------------------------------- # These lock the CURRENT worker/manager wire contract before any extraction # of kernel/manager internals. They follow the existing Kernel(...) patterns # exactly — do not add new frame types or reader loops here. def _contract_dispatcher(method, args): if method == "soft": # single-key {"error": ...} is the soft-fail shape: the manager must # route it onto the error channel, NOT hand it back as data. return {"error": "soft failure from host"} if method == "error_plus_data": # an error key WITH siblings is ordinary data, not a soft-fail. return {"error": "x", "detail": "still data"} if method == "none": return None raise ValueError(f"unknown method {method}") def test_response_frame_shape(): """The final response frame carries exactly the documented key set. `cwd` is a host-side annotation added by the manager, not a field the worker produces: the observation formatter needs a workspace-relative place to spill an oversized stdout, and the manager is the only layer that knows where that is. ``files_read`` is worker-observed response metadata; older managers ignore unknown response fields and remain wire-compatible. """ with Kernel(dispatcher=_echo_dispatcher) as k: r = k.execute("print('shape')") assert set(r) == { "type", "id", "stdout", "stderr", "error", "interrupted", "trace", "guards", "files_read", "usage", "cwd", } assert r["type"] == "response" assert r["interrupted"] is False assert set(r["trace"]) == {"error_lineno", "error_call"} assert isinstance(r["guards"], dict) def test_stderr_captured_separately(): with Kernel(dispatcher=_echo_dispatcher) as k: r = k.execute("import sys\nsys.stderr.write('warn!\\n')\nprint('out')") assert r["stdout"].strip() == "out" assert "warn!" in r["stderr"] assert r["error"] is None def test_stdout_chunks_stream_via_on_chunk(): """stdout_chunk frames stream live; on_chunk sees the same text the final response frame reports.""" chunks = [] with Kernel(dispatcher=_echo_dispatcher) as k: r = k.execute("print('live')", on_chunk=chunks.append) assert "live" in "".join(chunks) assert r["stdout"] == "live\n" def test_explicit_cell_id_roundtrips_through_worker_response(): with Kernel(dispatcher=_echo_dispatcher) as kernel: result = kernel.execute("print('identified')", cell_id="cell-shared") automatic_one = kernel.execute("pass") automatic_two = kernel.execute("pass") assert result["id"] == "cell-shared" assert result["stdout"] == "identified\n" assert automatic_one["id"] assert automatic_one["id"] != automatic_two["id"] def test_save_artifact_host_call_carries_canonical_and_declared_cell_ids(): calls = [] def dispatcher(method, args): calls.append((method, args)) return {"ok": True} with Kernel(dispatcher=dispatcher) as kernel: inherited = kernel.execute( "print(host.save_artifact('result.csv')['ok'])", cell_id="cell-artifact", ) explicit = kernel.execute( "host.save_artifact('result.csv', producing_cell_id='manual-cell')", cell_id="cell-other", ) assert inherited["error"] is None assert inherited["stdout"].strip() == "True" assert explicit["error"] is None save_calls = [call for call in calls if call[0] == "save_artifact"] assert save_calls == [ ( "save_artifact", [ { "path": "result.csv", "inputVersionIds": [], "priority": 0, "executionCellId": "cell-artifact", "producingCellId": "cell-artifact", } ], ), ( "save_artifact", [ { "path": "result.csv", "inputVersionIds": [], "producingCellId": "manual-cell", "priority": 0, "executionCellId": "cell-other", } ], ), ] def _require_matplotlib_in_the_kernel() -> None: # `find_spec`, not `importorskip`: importing matplotlib into the test # process is exactly the cost these tests are about, and not theirs to pay. import importlib.util if importlib.util.find_spec("matplotlib") is None: pytest.skip("matplotlib is not installed") _MATPLOTLIB_MODULES = ( "matplotlib", "matplotlib.figure", "matplotlib.font_manager", "matplotlib.pyplot", ) @pytest.mark.parametrize( ("mode", "guards_off"), [("repl", False), ("repl", True), ("jupyter", False)], ids=["guards-and-provenance", "provenance-only", "guards-only"], ) def test_a_cell_that_never_plots_never_loads_matplotlib( tmp_path, monkeypatch, mode, guards_off ): """Importing matplotlib builds its font list, and under an enforced sandbox every Kernel gets an empty private MPLCONFIGDIR, so that import was a font scan (`system_profiler` on macOS) of 8-48 seconds on the first Cell of every new kernel -- `x = 1` included. Two independent layers paid it for Cells that never plot: the figure-leak guard imported pyplot to read its figure numbers, and provenance imported `matplotlib.figure` to wrap `savefig`. Each parameter isolates one of them.""" _require_matplotlib_in_the_kernel() if guards_off: monkeypatch.setenv("OPENAI4S_GUARDS_OFF", "1") else: monkeypatch.delenv("OPENAI4S_GUARDS_OFF", raising=False) monkeypatch.delenv("OPENAI4S_PROVENANCE_OFF", raising=False) with Kernel(dispatcher=_echo_dispatcher, cwd=str(tmp_path), mode=mode) as kernel: assert kernel.execute("x = 1")["error"] is None probe = kernel.execute( "import sys\n" f"print([name for name in {_MATPLOTLIB_MODULES!r} if name in sys.modules])" ) assert probe["error"] is None assert probe["stdout"].strip() == "[]" def _figure_lineage_dispatcher(records): def dispatcher(method, args): if method == "prov_record": records.append(args[0]) return {"ok": True} if method == "prov_resolve_path": return None return _echo_dispatcher(method, args) return dispatcher _TAGGED_FIGURE_SAVE = ( "from openai4s.kernel import provenance\n" "figure = plt.figure()\n" "provenance.set_tags(figure, frozenset({'input-version'}))\n" "figure.savefig('figure.png')\n" "plt.close(figure)\n" ) @pytest.mark.parametrize("same_cell", [True, False], ids=["same-cell", "later-cell"]) def test_figure_savefig_lineage_survives_a_matplotlib_imported_after_install( tmp_path, same_cell ): """Provenance no longer imports matplotlib to wrap `Figure.savefig`; it wraps it the moment `matplotlib.figure` is imported. A wrapper applied only between Cells would miss the commonest plotting Cell there is -- import, plot and save in one go -- so both orders must report the edge.""" _require_matplotlib_in_the_kernel() records: list[dict] = [] importing = ( "import matplotlib\nmatplotlib.use('Agg')\nimport matplotlib.pyplot as plt\n" ) with Kernel( dispatcher=_figure_lineage_dispatcher(records), cwd=str(tmp_path) ) as kernel: assert kernel.execute("x = 1")["error"] is None if same_cell: saved = kernel.execute(importing + _TAGGED_FIGURE_SAVE) else: assert kernel.execute(importing)["error"] is None saved = kernel.execute(_TAGGED_FIGURE_SAVE) assert saved["error"] is None assert (tmp_path / "figure.png").is_file() figure_records = [ record for record in records if record.get("filename") == "figure.png" ] assert figure_records, records assert figure_records[-1]["input_version_ids"] == ["input-version"] def test_host_call_soft_fail_single_key_error_dict(): """Dispatcher returning {'error': msg} (and nothing else) surfaces in the kernel as a RuntimeError('host. error: ') — the soft-fail contract every host handler relies on.""" with Kernel(dispatcher=_contract_dispatcher) as k: r = k.execute( "try:\n" " host._call('soft', [])\n" "except RuntimeError as e:\n" " print('caught:', e)" ) assert r["error"] is None assert "caught: host.soft error: soft failure from host" in r["stdout"] def test_host_call_error_key_with_siblings_is_plain_data(): with Kernel(dispatcher=_contract_dispatcher) as k: r = k.execute( "d = host._call('error_plus_data', [])\n" "print(sorted(d), d['error'], d['detail'])" ) assert r["error"] is None assert r["stdout"].strip() == "['detail', 'error'] x still data" def test_host_call_none_data_roundtrips(): with Kernel(dispatcher=_contract_dispatcher) as k: r = k.execute("print(host._call('none', []) is None)") assert r["stdout"].strip() == "True" def test_host_call_without_dispatcher_errors(): with Kernel(dispatcher=None) as k: r = k.execute( "try:\n" " host._call('ping', [])\n" "except RuntimeError as e:\n" " print('caught:', e)" ) assert "no host dispatcher configured" in r["stdout"] def test_system_exit_is_trapped_and_worker_survives(): """exit()/SystemExit must not kill the worker: it is reported as an error and the SAME kernel (same namespace) keeps serving cells.""" with Kernel(dispatcher=_echo_dispatcher) as k: k.execute("x = 7") r = k.execute("raise SystemExit(3)") assert r["error"] is not None assert "SystemExit trapped" in r["error"] assert k.is_alive() r2 = k.execute("print(x)") # namespace survived the trapped exit assert r2["stdout"].strip() == "7" def test_restart_bumps_generation_and_resets_namespace(): with Kernel(dispatcher=_echo_dispatcher) as k: assert k.generation == 0 k.execute("x = 1") k.restart() assert k.generation == 1 assert k.is_alive() r = k.execute("print('x' in globals())") assert r["stdout"].strip() == "False" # --- inner-RPC wire contracts (PR 06/PR 10 reviewers) ----------------------- # 15MB wire cap, host_ack pre-response frames, bounded-discard desync # recovery, and the SIGINT interrupt contract. Extra pre-response frames are # injected through the manager's OWN _send/_service_host_call machinery — # the single-frame-reader loop and the host-call transaction stay intact. def test_host_call_wire_cap_rejects_oversized_payload(): """A host_call whose JSON frame exceeds the 15MB wire cap raises ValueError IN the kernel before anything is written — the dispatcher never sees the call and the channel stays usable for the next RPC.""" with Kernel(dispatcher=_echo_dispatcher) as k: r = k.execute( "big = 'x' * 15_000_001\n" "try:\n" " host._call('ping', [big])\n" "except ValueError as e:\n" " print('capped:', '15MB wire cap' in str(e))\n" "print(host._call('ping', []))" ) assert r["error"] is None # if the frame had reached the wire, 'ping' would have succeeded and # the 'capped:' line would be missing entirely assert r["stdout"].splitlines() == ["capped: True", "pong"] def test_host_ack_and_bounded_discard_recovery(): """Pre-response frames the worker must survive: a correct-id host_ack is skipped WITHOUT counting against the discard budget, and up to exactly _DISCARD_BUDGET (8) out-of-order/garbage frames are discarded before the id-matched response — the RPC still returns its data.""" with Kernel(dispatcher=_echo_dispatcher) as k: orig = k._service_host_call def noisy(frame): # correct-id ack: the "keep waiting" signal, never a discard k._send({"type": "host_ack", "id": frame["id"]}) # exactly 8 discardable frames = the full budget: # 6 stale responses (id mismatch) + 1 non-JSON line + 1 non-dict for i in range(6): k._send({"type": "host_response", "id": f"stale-{i}", "data": "old"}) k._proc.stdin.write("this line is not json\n") k._proc.stdin.flush() k._send([1, 2, 3]) orig(frame) k._service_host_call = noisy r = k.execute("print(host._call('ping', []))") assert r["error"] is None assert r["stdout"].strip() == "pong" def test_host_call_desync_over_budget_raises_and_kernel_survives(): """One frame past the discard budget (9 mismatched frames, no real response) makes host_call give up with a 'protocol desync' RuntimeError instead of spinning forever — and the worker keeps serving afterwards.""" with Kernel(dispatcher=_echo_dispatcher) as k: orig = k._service_host_call def flood(frame): for i in range(9): k._send({"type": "host_response", "id": f"stale-{i}", "data": None}) # never send the real response — the worker must bail on its own k._service_host_call = flood r = k.execute( "try:\n" " host._call('ping', [])\n" "except RuntimeError as e:\n" " print('caught:', e)" ) assert r["error"] is None assert "host.ping: protocol desync" in r["stdout"] # with a sane host again, the same worker answers the next RPC k._service_host_call = orig r2 = k.execute("print(host._call('ping', []))") assert r2["stdout"].strip() == "pong" def test_sigint_interrupt_reports_interrupted_true_lineno_none(): """The host.exec_interrupt contract: a DELIVERED SIGINT ends the cell with interrupted=True, error='Interrupted' and NO error_lineno — and the kernel (with its namespace) survives the interrupt. The gate is a host_call, not a stdout chunk. A chunk proved nothing it was once read as proving: `sys.stdout` is swapped to the chunk-emitting buffer at the top of the cell, well before user code runs and before the SIGINT handler used to be armed, and the guard phase in between has been measured at 18 seconds against a cold Matplotlib font cache. So a test that interrupted on the first chunk could deliver its one signal into the window where the worker swallowed it — and then fail, intermittently, on a Linux runner under load, blaming the interrupt path for a race in its own premise. A `host_call` frame can only be emitted from inside the cell's `host._call`, so receiving one is proof that `exec` is running user code. """ reached_user_code = threading.Event() def dispatcher(method, args): if method == "ping": reached_user_code.set() return _echo_dispatcher(method, args) with Kernel(dispatcher=dispatcher) as k: k.execute("marker = 'still-here'") result = {} def run(): result["r"] = k.execute( "host._call('ping', [])\nimport time\ntime.sleep(30)" ) t = threading.Thread(target=run, daemon=True) t.start() # 60s, not 15: a completely cold optional-science install can spend # most of a minute in the guard phase before the first user byte runs. assert reached_user_code.wait(60), "the cell never reached its host call" k.interrupt() t.join(timeout=30) assert not t.is_alive(), _interrupt_diagnosis(k) r = result["r"] assert r["interrupted"] is True assert r["error"] == "Interrupted" assert r["trace"]["error_lineno"] is None assert k.is_alive() assert k.execute("print(marker)")["stdout"].strip() == "still-here" def test_interrupt_stops_a_cell_whose_helper_threads_could_eat_the_signal(): """A worker with unblocked helper threads must still stop promptly. The worker is not single-threaded in practice: the guard phase's matplotlib import starts OpenBLAS's pool, and those threads inherit an empty signal mask. Linux may hand a process-directed SIGINT to any of them; CPython's Python-level handler runs only on the main thread, and a main thread blocked in `time.sleep` is never woken by a flag another thread set. Observed on a CI runner (run 32735586388, round 15): every thread with SigPnd 0, the main thread parked in hrtimer_nanosleep, and the cell reporting interrupted=True only at wall=30.0005 -- the stop was consumed by a BLAS thread and did nothing until the sleep ran out. The interrupt path now aims tgkill at the main thread, which this asserts with explicitly spawned stand-ins for the BLAS pool. """ reached = threading.Event() def dispatcher(method, args): if method == "ping": reached.set() return _echo_dispatcher(method, args) with Kernel(dispatcher=dispatcher) as k: result = {} def run(): result["r"] = k.execute( "import threading, time\n" "for _ in range(6):\n" " threading.Thread(target=time.sleep, args=(60,), " "daemon=True).start()\n" "host._call('ping', [])\n" "time.sleep(60)" ) t = threading.Thread(target=run, daemon=True) t.start() assert reached.wait(60), "the cell never reached its host call" k.interrupt() # Well under the cell's own 60s sleep: a signal eaten by a helper # thread leaves the sleep running and fails here, rather than being # indistinguishable from a slow runner at the sleep's full length. t.join(timeout=15) assert not t.is_alive(), _interrupt_diagnosis(k) r = result["r"] assert r["interrupted"] is True assert r["error"] == "Interrupted" def test_tgkill_helper_reaches_the_calling_threads_own_process(): """The syscall-number table and argument order, proven by delivery. On Linux the helper signals THIS process's main thread with SIGUSR1 and the installed handler must observe it -- a wrong syscall number or a swapped tgid/tid argument fails here, not in a hung CI cell. Elsewhere the helper must decline so the caller keeps the process-directed path. """ import sys as _sys if _sys.platform != "linux": assert ( manager_mod._signal_worker_main_thread(os.getpid(), signal.SIGUSR1) is False ) return seen = threading.Event() previous = signal.signal(signal.SIGUSR1, lambda s, f: seen.set()) try: assert manager_mod._signal_worker_main_thread( os.getpid(), signal.SIGUSR1 ), "tgkill failed on a platform whose syscall number is in the table" assert seen.wait(5), "tgkill reported success but the signal never arrived" finally: signal.signal(signal.SIGUSR1, previous) def _interrupt_diagnosis(kernel) -> str: """What a failing interrupt looked like, instead of `assert not True`. This assertion has failed twice in CI and said nothing either time, so the next occurrence carries its own evidence: whether the worker is alive, and what the kernel itself says about the signal. On Linux `/proc//status` settles the question this test cannot otherwise answer — `SigCgt` bit 1 set means the worker HAS a SIGINT handler installed, `SigIgn` means it is ignoring the signal, and `ShdPnd` means one was delivered and never taken. """ lines = ["interrupt did not stop the cell"] proc = getattr(kernel, "_proc", None) pid = getattr(proc, "pid", None) lines.append(f" worker pid={pid} alive={kernel.is_alive()}") if pid is not None: try: status = Path(f"/proc/{pid}/status").read_text(encoding="utf-8") except OSError: lines.append(" /proc unavailable (not Linux, or the worker is gone)") else: wanted = ("State", "SigIgn", "SigCgt", "SigPnd", "ShdPnd") for line in status.splitlines(): if line.split(":", 1)[0] in wanted: lines.append(f" {line.strip()}") lines.append(" (SIGINT is bit 1: mask 0x2 in the Sig* words)") return "\n".join(lines) @pytest.mark.skipif(os.name != "posix", reason="POSIX signal dispositions") def test_kernel_children_start_with_default_sigint_even_when_spawner_ignores_it(): """The spawn boundary undoes an inherited SIG_IGN on SIGINT/SIGQUIT. A shell backgrounds `./start.sh &` (and nohup) with both set to SIG_IGN, which survives every exec: daemon → [bwrap] → sh → Rscript. R installs its interrupt handler only when the inherited disposition is not SIG_IGN, so a backgrounded daemon's R kernels silently dropped every Kernel.interrupt() while the python worker — which calls signal.signal unconditionally — kept working. PipeTransport is the one local spawn site for both, so the probe here covers Python, R and the bwrap-wrapped argv alike. The probe reads what CPython concluded from the inherited disposition: `default_int_handler` installed means the child was born with SIGINT not ignored (the reset worked); SIG_IGN means the legacy leaked through. The second spawn runs on a non-main thread because that is where the daemon actually spawns kernels — it pins that the preexec reset (which calls `signal.signal` in the forked child) stays legal off-main-thread. """ import sys from openai4s.kernel.transport import PipeTransport probe = ( "import signal, sys\n" "sys.stdout.write('int_reset=%s quit_reset=%s\\n' % (\n" " signal.getsignal(signal.SIGINT) is signal.default_int_handler,\n" " signal.getsignal(signal.SIGQUIT) is signal.SIG_DFL))\n" ) command = [sys.executable, "-c", probe] def spawn_and_read() -> str: transport = PipeTransport(command, cwd=None, env=dict(os.environ)) try: line = transport.read_line() transport.process.wait(timeout=10) finally: transport.close(graceful=False) return line.strip() old_int = signal.signal(signal.SIGINT, signal.SIG_IGN) old_quit = signal.signal(signal.SIGQUIT, signal.SIG_IGN) try: assert spawn_and_read() == "int_reset=True quit_reset=True" box: dict[str, str] = {} thread = threading.Thread( target=lambda: box.update(line=spawn_and_read()), daemon=True ) thread.start() thread.join(30) assert not thread.is_alive(), "off-main-thread spawn never finished" assert box["line"] == "int_reset=True quit_reset=True" finally: signal.signal(signal.SIGINT, old_int) signal.signal(signal.SIGQUIT, old_quit) def test_a_chunk_from_another_cell_is_not_this_cell_s_output(monkeypatch): """`on_chunk` and the assembled stdout belong to the cell that was asked for. A frame stamped with a different cell id used to satisfy both — so a `logging.StreamHandler` still bound to a finished cell's `sys.stdout`, or a background thread writing after its cell returned, fed text into the next cell's stream and told a live watcher that user code had started. The interrupt contract is one of the things that watcher decides: a host that stops a cell on its first output would have been aiming at a cell that had not begun.""" seen: list[str] = [] with Kernel(dispatcher=_echo_dispatcher) as k: real_readline = k._readline injected = {"done": False} def readline_once_from_another_cell(): if not injected["done"]: injected["done"] = True return { "type": "stdout_chunk", "id": "a-cell-that-is-not-this-one", "text": "not mine\n", } return real_readline() monkeypatch.setattr(k, "_readline", readline_once_from_another_cell) out = k.execute("print('mine')", on_chunk=seen.append) assert injected["done"], "the foreign frame was never injected" assert "not mine" not in "".join(seen), "a foreign cell's text reached on_chunk" assert "not mine" not in out["stdout"] assert out["stdout"].strip() == "mine" assert k._stale_stdout_chunks == 1, "the dropped frame was not counted" def test_user_raised_keyboardinterrupt_is_normal_error_with_lineno(): """The contrast half of the SIGINT contract: user code raising KeyboardInterrupt itself is a NORMAL error (interrupted=False) with a real lineno — only a delivered signal sets interrupted=True.""" with Kernel(dispatcher=_echo_dispatcher) as k: r = k.execute("x = 1\nraise KeyboardInterrupt('manual')") assert r["interrupted"] is False assert "KeyboardInterrupt" in r["error"] assert r["trace"]["error_lineno"] == 2 assert k.is_alive() @pytest.fixture def sigint_probe(monkeypatch): """Observe which SIGINT handler the worker installs, without installing one. The contract under test is *which* handler is armed at each moment, so the probe records the installs rather than performing them: a test that really re-pointed this process's SIGINT would be asserting the worker's behaviour by adopting it. """ import signal as signal_mod installed = {"handler": None} def fake_signal(signum, handler): assert signum == signal_mod.SIGINT previous = installed["handler"] installed["handler"] = handler return previous monkeypatch.setattr(worker_mod.signal, "signal", fake_signal) for cell in ( worker_mod._in_user_code, worker_mod._sigint_delivered, worker_mod._sigint_pending, ): cell[0] = False yield installed for cell in ( worker_mod._in_user_code, worker_mod._sigint_delivered, worker_mod._sigint_pending, ): cell[0] = False def test_a_sigint_that_beats_user_code_is_owed_not_dropped(sigint_probe): """`Kernel.interrupt()` sends exactly ONE signal, and `_arm_sigint` runs before the cell is compiled -- so a stop pressed a millisecond early lands while the handler cannot raise. Swallowing it there (and disarming, which is what the handler used to do) made the rest of the cell uninterruptible: the user saw no interrupt, no error, and a cell that ran to completion.""" worker_mod._arm_sigint() assert sigint_probe["handler"] is worker_mod._sigint_handler worker_mod._sigint_handler(worker_mod.signal.SIGINT, None) # must not raise assert worker_mod._sigint_pending[0] is True, "the signal was dropped" assert ( sigint_probe["handler"] is worker_mod._sigint_handler ), "the handler disarmed itself; the rest of the cell cannot be interrupted" worker_mod._in_user_code[0] = True with pytest.raises(KeyboardInterrupt): worker_mod._raise_if_sigint_pending() # Reported as a DELIVERED signal, so the cell ends interrupted=True with no # error_lineno -- not as user code raising KeyboardInterrupt itself. assert worker_mod._sigint_delivered[0] is True assert worker_mod._sigint_pending[0] is False assert sigint_probe["handler"] is worker_mod._sigint_swallow, "not one-shot" def test_arming_a_cell_clears_a_signal_owed_to_the_previous_one(sigint_probe): """A pending flag that outlived its cell would interrupt the next one at its first bytecode, for a stop the user pressed against a cell that has already finished.""" worker_mod._sigint_pending[0] = True worker_mod._sigint_delivered[0] = True worker_mod._arm_sigint() assert worker_mod._sigint_pending[0] is False assert worker_mod._sigint_delivered[0] is False worker_mod._in_user_code[0] = True worker_mod._raise_if_sigint_pending() # nothing owed: must not raise def test_a_sigint_during_a_protocol_write_still_finishes_the_frame(monkeypatch): """A frame is written and flushed as one thing, or it is not a frame. `write` fills a buffer; `flush` is what reaches the host. A KeyboardInterrupt raised between them leaves a partial line on the channel, and the next flush concatenates it with the frame that follows -- `Kernel._readline` hands the result to `json.loads`, so a correctly handled interrupt surfaces to the caller as a JSONDecodeError from a stream that no longer parses. A cell's stdout goes out through exactly this path, which is where a stop lands. """ import signal as signal_mod import sys as _sys import threading as _threading if not hasattr(_sys, "_openai4s_protocol_lock"): monkeypatch.setattr( _sys, "_openai4s_protocol_lock", _threading.Lock(), raising=False ) monkeypatch.setattr(worker_mod.signal, "signal", lambda *_args: None) calls: list[str] = [] class _SignallingSink: """The protocol channel, with a SIGINT arriving mid-frame.""" def write(self, text): calls.append("write") worker_mod._sigint_handler(signal_mod.SIGINT, None) def flush(self): calls.append("flush") monkeypatch.setattr(worker_mod, "_proto_out", lambda: _SignallingSink()) worker_mod._in_user_code[0] = True worker_mod._sigint_delivered[0] = False worker_mod._sigint_pending[0] = False try: with pytest.raises(KeyboardInterrupt): worker_mod._write_frame({"type": "stdout_chunk", "id": "c1", "text": "hi"}) assert calls == ["write", "flush"], ( "the interrupt escaped before the frame was flushed; the host's next " f"read gets half a line (calls: {calls})" ) lock = worker_mod._write_lock() acquired = lock.acquire(blocking=False) try: assert acquired, "the interrupt left the protocol write lock held" finally: if acquired: lock.release() # Deferred, not swallowed: the cell still ends as interrupted. assert worker_mod._sigint_delivered[0] is True finally: for cell in ( worker_mod._in_user_code, worker_mod._sigint_delivered, worker_mod._sigint_pending, ): cell[0] = False def test_an_interrupt_inside_a_host_call_leaves_the_kernel_usable(): """The seam the deferred-signal fix opens, asserted rather than assumed. A protocol write now finishes before a latched SIGINT is raised, so the interrupt can land in a new place: between a `host_call` request going out and its `host_response` being read. The cell abandons an RPC the host has already dispatched, and the answer arrives for nobody — it reaches the worker's main loop as a frame that is not a request. That is contained by the loop's bounded discard, but "contained" is a claim, and an abandoned RPC that poisoned the next cell would look exactly like a healthy kernel until someone made a second host call. """ interrupted_from_the_dispatcher = threading.Event() with Kernel(dispatcher=_echo_dispatcher) as k: def dispatcher(method, args): if method == "ping" and not interrupted_from_the_dispatcher.is_set(): interrupted_from_the_dispatcher.set() # Delivered while the worker is blocked reading this call's # response — the exact window the deferral creates. k.interrupt() return _echo_dispatcher(method, args) k.dispatcher = dispatcher k.execute("marker = 'still-here'") result = k.execute("host._call('ping', [])\nimport time\ntime.sleep(30)") assert interrupted_from_the_dispatcher.is_set(), "the host call never ran" assert result["interrupted"] is True, result assert k.is_alive() assert k.execute("print(marker)")["stdout"].strip() == "still-here" # The abandoned RPC must not have consumed the next one's answer. assert k.execute("print(host._call('ping', []))")["stdout"].strip() == "pong" def test_a_worker_diagnostic_is_retained_rather_than_dropped(monkeypatch): """`log` frames were read and discarded, so the worker had no way to tell the host anything that does not fit in a response. It needs one: a cell whose SIGINT handler could not be armed cannot be stopped at all, and looks exactly like a slow cell until something says otherwise.""" with Kernel(dispatcher=_echo_dispatcher) as k: real_readline = k._readline injected = {"done": False} def readline_once_with_a_diagnostic(): if not injected["done"]: injected["done"] = True return {"type": "log", "msg": "SIGINT could not be armed for this cell"} return real_readline() monkeypatch.setattr(k, "_readline", readline_once_with_a_diagnostic) k.execute("print('ok')") assert injected["done"] assert any("SIGINT could not be armed" in line for line in k.worker_log_tail) def test_arming_failure_is_announced_instead_of_returning_silently(monkeypatch): """Off the main thread `signal.signal` refuses, and `_arm_sigint` returned normally anyway -- so the cell ran under the previous cell's swallow handler with `_sigint_delivered` already cleared. Every stop discarded, and the response frame reporting `interrupted: False` as though none had been asked for.""" frames: list[dict] = [] monkeypatch.setattr(worker_mod, "_write_frame", frames.append) def refuse(*_args): raise ValueError("signal only works in main thread of the main interpreter") monkeypatch.setattr(worker_mod.signal, "signal", refuse) assert worker_mod._arm_sigint() is False assert frames, "the failure was silent" assert frames[0]["type"] == "log" assert "could not be armed" in frames[0]["msg"] assert "watchdog" in frames[0]["msg"], "say what does end the cell instead" @pytest.mark.skipif(os.name != "posix", reason="process sessions are POSIX") def test_the_worker_runs_in_its_own_session(): """A signal aimed at the daemon's process group must not also be aimed at every cell running under it. ``PipeTransport`` owns this session for both the direct worker and the bubblewrap wrapper path. The latter must not create a second, nested session or its Cell children become unreachable to the watchdog's group stop.""" with Kernel(dispatcher=_echo_dispatcher) as k: worker_pid = k._proc.pid assert os.getpgid(worker_pid) == worker_pid, "the worker leads no group" assert os.getpgid(worker_pid) != os.getpgid(os.getpid()) @pytest.mark.skipif(os.name != "posix", reason="process groups are POSIX") def test_killing_the_worker_also_ends_what_the_cell_started(): """The kernel was the one long-lived child here with no group-scoped stop. `proc.kill()` ends the leader; a cell's own subprocess is a grandchild and survived it, so a watchdog kill left the actual work running with nothing holding a handle to it. The group stop was not merely missing -- it was unaddressable, because `os.getpgid(worker)` WAS the daemon's group and signalling it would have taken the daemon down. Session isolation is what makes the ladder pointable, so the two land together. Deliberately a raw `subprocess.Popen`, not `host.bash`: bash already puts itself in its own session, so it would prove nothing about the worker's. """ with Kernel(dispatcher=_echo_dispatcher) as k: result = k.execute( "import subprocess, sys\n" "child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)'])\n" "print(child.pid)" ) assert result["error"] is None, result grandchild = int(result["stdout"].strip()) worker_pidfd = getattr(k._sandbox, "_bwrap_worker_pidfd", None) if worker_pidfd is not None: # WSL/team kernels report namespace-local PIDs. Resolve the sole # child through the pinned worker before inspecting it on the Host. fields = dict( line.split(":", 1) for line in Path(f"/proc/self/fdinfo/{worker_pidfd}") .read_text() .splitlines() if ":" in line ) worker = int(fields["Pid"]) children = ( Path(f"/proc/{worker}/task/{worker}/children").read_text().split() ) assert len(children) == 1 host_child = int(children[0]) status = Path(f"/proc/{host_child}/status").read_text() ns_pid = next( line for line in status.splitlines() if line.startswith("NSpid:") ) assert int(ns_pid.split()[-1]) == grandchild grandchild = host_child assert os.getpgid(grandchild) == os.getpgid(k._proc.pid), ( "the cell's subprocess is not in the worker's group, so this test " "would pass without the stop ladder ever being exercised" ) k.kill_worker() deadline = time.time() + 10.0 while time.time() < deadline: try: os.kill(grandchild, 0) except ProcessLookupError: break time.sleep(0.05) else: # pragma: no cover - the failure this test exists for try: os.kill(grandchild, signal.SIGKILL) except OSError: pass raise AssertionError( f"the cell's subprocess {grandchild} outlived the worker; " "killing the leader left the actual work running" ) def test_an_idle_worker_survives_an_interrupt_before_its_first_cell(): """Interrupt stops a CELL. It must never end the worker. Until the first cell armed a handler the worker kept Python's default SIGINT disposition, so a stop delivered to an idle kernel raised KeyboardInterrupt straight out of its own read loop and took the namespace with it. `inspect_variables` is what proves the worker has reached that loop, so this asserts the contract instead of racing it. """ with Kernel(dispatcher=_echo_dispatcher) as k: k.inspect_variables(limit=1) # the worker is in its read loop k.interrupt() assert k.is_alive() result = k.execute("marker = 'survived'\nprint(marker)") assert result["error"] is None assert result["stdout"].strip() == "survived" def test_host_bash_is_kernel_local_and_never_rpcs(tmp_path): """host.bash runs INSIDE the worker process — the host executes only python/R cells. A dispatcher that rejects a 'bash' method proves no RPC happens; the command's output is captured in the cell like any subprocess.""" authorized = _authorized_bash_dispatcher(tmp_path) def no_bash_dispatcher(method, args): if method == "bash": raise AssertionError("host.bash must not reach the host dispatcher") return authorized(method, args) with Kernel(dispatcher=no_bash_dispatcher, cwd=str(tmp_path)) as k: r = k.execute("r = host.bash('echo kernel-local'); print(r['stdout'])") assert r["error"] is None assert "kernel-local" in r["stdout"] # the shell ran in the worker's cwd (the workspace) r2 = k.execute("print(host.bash('pwd')['workdir'])") assert str(tmp_path.resolve()) in r2["stdout"] methods = [ row[0] for row in authorized.store._conn.execute( "SELECT method FROM host_call_log ORDER BY created_at" ).fetchall() ] assert "bash" in methods # safe result audit assert "authorize_bash" in methods def test_host_bash_capability_is_bound_to_exact_worker_spawn(tmp_path): dispatcher = _authorized_bash_dispatcher(tmp_path) observed: list[tuple[str, str | int | None]] = [] authorize = dispatcher._bash_authorization.authorize def inspect_binding(spec): observed.append((spec.get("generation"), dispatcher.bash_generation_id)) return authorize(spec) dispatcher._bash_authorization.authorize = inspect_binding with Kernel(dispatcher=dispatcher, cwd=str(tmp_path)) as kernel: first_generation = kernel.authorization_generation first = kernel.execute("print(host.bash('echo first')['stdout'])") assert first["error"] is None kernel.restart() second_generation = kernel.authorization_generation second = kernel.execute("print(host.bash('echo second')['stdout'])") assert second["error"] is None assert first_generation != second_generation assert observed == [ (first_generation, first_generation), (second_generation, second_generation), ] assert dispatcher.bash_generation_id is None def test_host_bash_static_precheck_blocks_catastrophe(tmp_path): with Kernel(dispatcher=_echo_dispatcher, cwd=str(tmp_path)) as k: r = k.execute( "try:\n" " host.bash('rm -rf /')\n" "except RuntimeError as e:\n" " print('BLOCKED:', e)\n" ) assert r["error"] is None assert "BLOCKED:" in r["stdout"] and "precheck" in r["stdout"] if __name__ == "__main__": raise SystemExit(pytest.main([__file__, "-v"])) def test_one_print_cannot_put_an_unbounded_frame_on_the_protocol_pipe(): """A real subprocess, because the defect is in the pipe, not the arithmetic. `_StreamingStdout.write` forwarded whatever string it was handed as one `stdout_chunk` frame, and `_write_frame` had no size gate at all -- the 15MB `_HOST_CALL_WIRE_CAP` guards only the inbound direction. So `print("x" * 200_000_000)` produced a single ~200MB JSON line, which the host's `readline()` then materialised whole: ~200MB allocated on both sides from one ordinary statement, with nothing in between to refuse it. Driven through `Kernel` so the frames really cross a pipe and a real `readline` really reads them. A `StringIO` here would prove the slicing and none of the thing that broke. """ chunks: list[str] = [] with Kernel(dispatcher=_echo_dispatcher) as k: result = k.execute( "print('x' * 20_000_000)", on_chunk=lambda text: chunks.append(text) ) assert result["error"] is None # Every frame that crossed the pipe is individually bounded. assert chunks, "the cell streamed nothing at all" assert max(len(chunk) for chunk in chunks) <= 64_000 + 64 # And the stream as a whole is bounded, with one marker rather than one # per oversized write. streamed = "".join(chunks) assert len(streamed) < 20_000_000 assert streamed.count("...(truncated at") == 1 # The captured result stays bounded too, and says what it counted. assert len(result["stdout"]) <= 1_000_000 + 64 assert "characters" in result["stdout"]