"""Gateway integration contracts for supervised Python/R kernel lifecycles.""" from __future__ import annotations import json import re import threading from pathlib import Path from types import SimpleNamespace import pytest from openai4s.config import Config, LLMConfig from openai4s.host.data import kernel_artifact_input_dir from openai4s.server import gateway as gateway_mod from openai4s.skills_loader.versions import project_skills_root class _Hub: def __init__(self) -> None: self.events: list[dict] = [] def emitter(self, root_frame_id: str): def emit(event: dict) -> None: event.setdefault("root_frame_id", root_frame_id) self.events.append(event) return emit def broadcast(self, root_frame_id: str, event: dict) -> None: event.setdefault("root_frame_id", root_frame_id) self.events.append(event) def has_subscriber(self, root_frame_id: str) -> bool: return False def _runner(tmp_path, *, team_mode: bool = False): cfg = Config( data_dir=tmp_path, llm=LLMConfig(provider="deepseek", api_key="test-key"), max_turns=3, ) cfg.team_mode = team_mode return gateway_mod.SessionRunner(cfg, _Hub()) class _RecordingKernel: def __init__(self, name: str, events: list[str] | None = None) -> None: self.name = name self.dispatcher = SimpleNamespace(last_output=None) self.events = events if events is not None else [] self.live = True self.interrupt_calls = 0 self.shutdown_calls = 0 self.restart_calls = 0 self.kill_calls = 0 self.execute_entered = threading.Event() self.execute_release = threading.Event() self.release_on_interrupt = False def is_alive(self) -> bool: return self.live def execute(self, code, origin="agent", on_chunk=None, *, cell_id=None): self.events.append(f"{self.name}:execute-enter") self.execute_entered.set() assert self.execute_release.wait(2) self.events.append(f"{self.name}:execute-exit") return {"stdout": "", "stderr": "", "error": None} def interrupt(self) -> None: self.interrupt_calls += 1 self.events.append(f"{self.name}:interrupt") if self.release_on_interrupt: self.execute_release.set() def shutdown(self) -> None: self.shutdown_calls += 1 self.events.append(f"{self.name}:shutdown") self.live = False def restart(self) -> None: self.restart_calls += 1 self.live = True def kill_worker(self) -> None: self.kill_calls += 1 self.live = False self.execute_release.set() def test_stop_interrupts_both_slots_then_waits_for_execution_barrier(tmp_path): runner = _runner(tmp_path) st = runner._state("frame-stop", "default") events: list[str] = [] python = _RecordingKernel("python", events) python.release_on_interrupt = True r = _RecordingKernel("r", events) st.kernels.ensure("python", "base", lambda: python) st.kernels.ensure("r", None, lambda: r) def execute_turn() -> None: with st.turn_lock: python.execute("pass") turn = threading.Thread(target=execute_turn) turn.start() assert python.execute_entered.wait(1) result = runner.stop_kernel(st.root_frame_id) turn.join(1) assert not turn.is_alive() assert result["state"] == "stopped" assert python.interrupt_calls == r.interrupt_calls == 1 assert events.index("python:interrupt") < events.index("python:execute-exit") assert events.index("python:execute-exit") < events.index("python:shutdown") assert runner.kernel_status(st.root_frame_id)["state"] == "stopped" # Stop keeps cancellation asserted until the next explicit start/turn. assert st.cancel.is_set() def test_stop_intent_cannot_be_overtaken_by_a_new_start(monkeypatch, tmp_path): runner = _runner(tmp_path) st = runner._state("frame-stop-race", "default") current = _RecordingKernel("current") st.kernels.ensure("python", "base", lambda: current) cancel_reached = threading.Event() let_stop_wait_for_barrier = threading.Event() original_cancel = runner._cancel_current_for_lifecycle def paused_cancel(root_frame_id: str, *, reason: str) -> None: original_cancel(root_frame_id, reason=reason) cancel_reached.set() assert let_stop_wait_for_barrier.wait(2) monkeypatch.setattr(runner, "_cancel_current_for_lifecycle", paused_cancel) replacement = _RecordingKernel("replacement") start_entered = threading.Event() def ensure(state) -> None: start_entered.set() state.kernels.ensure("python", "base", lambda: replacement) monkeypatch.setattr(runner, "_ensure_kernel", ensure) stop_result = {} start_result = {} start_attempted = threading.Event() stopping = threading.Thread( target=lambda: stop_result.update(runner.stop_kernel(st.root_frame_id)) ) stopping.start() assert cancel_reached.wait(1) def start() -> None: start_attempted.set() start_result.update(runner.start_kernel(st.root_frame_id)) starting = threading.Thread(target=start) starting.start() assert start_attempted.wait(1) assert not start_entered.wait(0.1) let_stop_wait_for_barrier.set() stopping.join(1) starting.join(1) assert not stopping.is_alive() and not starting.is_alive() assert stop_result["state"] == "stopped" assert start_result["state"] == "running" assert current.shutdown_calls == 1 assert st.kernel is replacement assert not st.cancel.is_set() lifecycle = [ event["status"] for event in runner.hub.events if event.get("type") == "kernel_status" ] assert lifecycle[-2:] == ["stopped", "started"] def test_ensure_replaces_a_dead_supervised_python_worker(monkeypatch, tmp_path): runner = _runner(tmp_path) st = runner._state("frame-dead", "default") old = _RecordingKernel("old") st.kernels.ensure("python", "base", lambda: old) old.live = False replacement = _RecordingKernel("replacement") calls = [] def spawn(state): calls.append("spawn") return state.kernels.ensure("python", "base", lambda: replacement) monkeypatch.setattr(runner, "_spawn_kernel", spawn) runner._ensure_kernel(st) assert calls == ["spawn"] assert st.kernel is replacement assert old.shutdown_calls == 1 def test_python_bootstrap_runs_outside_supervisor_lock(monkeypatch, tmp_path): runner = _runner(tmp_path) st = runner._state("frame-bootstrap", "default") st.messages = [{"role": "system", "content": "test"}] env = SimpleNamespace( name="base", interpreter="base-python", root=tmp_path / "base", is_conda=False, bin_dir=None, ) monkeypatch.setattr( runner, "_resolve_env", lambda state: (setattr(state, "env_name", "base") or env), ) monkeypatch.setattr(runner, "_wire_delegation", lambda *args, **kwargs: None) monkeypatch.setattr( gateway_mod, "build_dispatcher", lambda *args, **kwargs: SimpleNamespace(active_r_env=None), ) runner.skills = SimpleNamespace(bootstrap_code="bootstrap()") class BlockingBootstrapKernel: instance = None created = threading.Event() def __init__(self, dispatcher, **kwargs) -> None: self.dispatcher = dispatcher self.live = True self.entered = threading.Event() self.release = threading.Event() BlockingBootstrapKernel.instance = self BlockingBootstrapKernel.created.set() def is_alive(self) -> bool: return self.live def execute(self, code, origin="agent", on_chunk=None): self.entered.set() assert self.release.wait(2) probe = re.search(r"__OPENAI4S_PY_ENV_[0-9a-f]+__", code) payload = { "runtime_version": "3.14-test", "interpreter": "base-python", "prefix": "/test-env", "base_prefix": "/test-env", "sdk_version": "0.1.0", "provenance_version": "1", "host_capability_version": "2", "package_manifest": [], "locale": {"preferred_encoding": "UTF-8"}, } return { "stdout": (probe.group(0) + json.dumps(payload)) if probe else "", "stderr": "", "error": None, } def interrupt(self) -> None: self.release.set() def shutdown(self) -> None: self.live = False monkeypatch.setattr(gateway_mod, "Kernel", BlockingBootstrapKernel) spawn_done = threading.Event() def spawn() -> None: with st.turn_lock: runner._spawn_kernel(st) spawn_done.set() spawning = threading.Thread(target=spawn) spawning.start() assert BlockingBootstrapKernel.created.wait(1) kernel = BlockingBootstrapKernel.instance assert kernel is not None assert kernel.entered.wait(1) interrupt_done = threading.Event() def interrupt() -> None: st.kernels.interrupt("python") interrupt_done.set() interrupting = threading.Thread(target=interrupt) interrupting.start() acquired_without_bootstrap_finishing = interrupt_done.wait(0.5) if not acquired_without_bootstrap_finishing: # Cleanup makes a regression fail promptly instead of hanging pytest. kernel.release.set() interrupting.join(1) spawning.join(1) assert acquired_without_bootstrap_finishing assert spawn_done.is_set() def test_environment_replacement_keeps_session_dispatcher_and_commits_active_env( monkeypatch, tmp_path ): from openai4s.kernel import environments as envmod runner = _runner(tmp_path) frame_id = runner.store.new_frame(kind="turn", project_id="default", status="ready") st = runner._state(frame_id, "default") st.messages = [{"role": "system", "content": "test"}] envs = { "base": SimpleNamespace( name="base", interpreter="base-python", root=tmp_path / "base", is_conda=False, bin_dir=str(tmp_path / "base" / "bin"), language="python", python_version=lambda: "3.14", ), "struct": SimpleNamespace( name="struct", interpreter="struct-python", root=tmp_path / "struct", is_conda=False, bin_dir=str(tmp_path / "struct" / "bin"), language="python", python_version=lambda: "3.14", ), } monkeypatch.setattr(envmod, "get_environment", envs.get) monkeypatch.setattr(envmod, "default_env_name", lambda: "base") dispatchers = [] class Dispatcher: def __init__(self) -> None: self.last_output = None self.active_r_env = None def __call__(self, method, args): return None def make_dispatcher(*args, **kwargs): dispatcher = Dispatcher() dispatchers.append(dispatcher) return dispatcher fail_struct = {"value": True} kernels = [] class FakeKernel: def __init__(self, dispatcher, python, **kwargs) -> None: if python == "struct-python" and fail_struct["value"]: raise RuntimeError("struct worker failed to start") self.dispatcher = dispatcher self.python = python self.options = kwargs self.live = True self.shutdown_calls = 0 kernels.append(self) def is_alive(self) -> bool: return self.live def shutdown(self) -> None: self.shutdown_calls += 1 self.live = False bootstrapped = [] monkeypatch.setattr(gateway_mod, "build_dispatcher", make_dispatcher) monkeypatch.setattr(gateway_mod, "Kernel", FakeKernel) monkeypatch.setattr(runner, "_wire_delegation", lambda *args, **kwargs: None) monkeypatch.setattr( runner, "_run_bootstrap", lambda state, kernel=None, workspace=None: bootstrapped.append(kernel), ) st.desired_env = "base" first = runner._spawn_kernel(st) old_kernel = first.kernel old_dispatcher = st.dispatcher old_dispatcher.active_r_env = "r-special" background = old_dispatcher.background_kernel_factory() assert background.python == "base-python" assert background.options["cwd"] == str(st.workspace) with pytest.raises(RuntimeError, match="failed to start"): runner.set_env(frame_id, "struct") assert st.kernel is old_kernel assert st.dispatcher is old_dispatcher assert st.env_name == "base" assert st.desired_env == "struct" assert old_kernel.shutdown_calls == 0 assert st.kernels.status("python")["generation"] == 0 fail_struct["value"] = False changed = runner.set_env(frame_id, "struct") assert changed["generation"] == 1 assert st.kernel is kernels[-1] and st.kernel is not old_kernel assert len(dispatchers) == 1 assert st.dispatcher is old_dispatcher assert st.kernel.dispatcher is old_dispatcher assert st.dispatcher.active_r_env == "r-special" replacement_background = st.dispatcher.background_kernel_factory() assert replacement_background.python == "struct-python" assert replacement_background.options["cwd"] == str(st.workspace) assert st.env_name == "struct" assert old_kernel.shutdown_calls == 1 assert bootstrapped == [old_kernel, st.kernel] def test_team_mode_composes_read_isolation_into_every_local_kernel( monkeypatch, tmp_path ): from openai4s.kernel import r_kernel as r_kernel_mod runner = _runner(tmp_path, team_mode=True) st = runner._state("frame-team-isolation", "default") st.messages = [{"role": "system", "content": "test"}] environment = SimpleNamespace( name="base", interpreter="base-python", root=tmp_path / "base", is_conda=False, bin_dir=None, ) monkeypatch.setattr( runner, "_resolve_env", lambda state: (setattr(state, "env_name", "base") or environment), ) monkeypatch.setattr(runner, "_run_bootstrap", lambda *_args, **_kwargs: {}) created = [] class FakeKernel: def __init__(self, dispatcher, **options) -> None: self.dispatcher = dispatcher self.options = options self.live = True created.append(self) def is_alive(self) -> bool: return self.live def shutdown(self) -> None: self.live = False monkeypatch.setattr(gateway_mod, "Kernel", FakeKernel) lease = runner._spawn_kernel(st) isolation_root = str(tmp_path.resolve()) policy = lease.kernel.options["read_isolation"] assert policy.roots == (isolation_root,) assert Path(policy.allowed_roots[0]) == kernel_artifact_input_dir( tmp_path, st.root_frame_id ) background = st.dispatcher.background_kernel_factory() assert background.options["read_isolation"].roots == (isolation_root,) assert st.delegation_runner.read_isolation.roots == (isolation_root,) r_options = [] def spawn_r_kernel(**options): r_options.append(options) return _RecordingKernel("r") monkeypatch.setattr(r_kernel_mod, "spawn_r_kernel", spawn_r_kernel) monkeypatch.setattr( gateway_mod, "bootstrap_r_generation", lambda _kernels, _workspace, _lease: {}, ) assert runner._ensure_r_kernel(st) is None assert r_options == [ { "cwd": str(st.workspace), "env": None, "read_isolation": st.delegation_runner.read_isolation, } ] def test_team_first_action_background_cell_gets_read_isolation(monkeypatch, tmp_path): runner = _runner(tmp_path, team_mode=True) st = runner._state("frame-team-background-first", "default") environment = SimpleNamespace( name="base", interpreter="base-python", root=tmp_path / "base", is_conda=False, bin_dir=None, ) monkeypatch.setattr(runner, "_resolve_env", lambda _state: environment) created = [] class FakeKernel: def __init__(self, dispatcher, **options) -> None: self.dispatcher = dispatcher self.options = options created.append(self) def execute(self, code, origin="agent", on_chunk=None): if on_chunk is not None: on_chunk("done") return {"stdout": "done", "stderr": "", "error": None} def shutdown(self) -> None: pass monkeypatch.setattr(gateway_mod, "Kernel", FakeKernel) dispatcher = runner._ensure_runtime(st) assert st.kernels.lease("python") is None launched = dispatcher._m_exec_background({"code": "print('first')"}) deadline = threading.Event() for _ in range(100): peek = dispatcher._m_exec_peek(launched["exec_id"]) if peek["done"]: break deadline.wait(0.01) else: # pragma: no cover - bounded thread scheduling failure raise AssertionError("background Cell did not finish") assert peek["error"] is None assert len(created) == 1 options = created[0].options assert Path(options["cwd"]) == st.local_workspace policy = options["read_isolation"] assert policy is not None assert policy.roots == (str(tmp_path.resolve()),) assert kernel_artifact_input_dir( runner.cfg.data_dir, st.root_frame_id ).resolve() in {Path(root).resolve() for root in policy.allowed_roots} assert st.kernels.lease("python") is None def test_team_read_policy_covers_data_personal_and_only_current_project_sidecars( monkeypatch, tmp_path, ): canonical_temp = tmp_path / "canonical-temp" canonical_temp.mkdir() monkeypatch.setattr(gateway_mod.tempfile, "gettempdir", lambda: str(canonical_temp)) runner = _runner(tmp_path / "daemon", team_mode=True) scratch = tmp_path / "scratch" scratch.mkdir() runner.cfg.data_roots = [scratch] st = runner._state("frame-team-owner", "default") owner = runner.store.team.create_user( username="alice", password="test-password-not-real" ) runner.store.team.set_session_owner( st.root_frame_id, owner["id"], project_id=st.project_id ) current = project_skills_root(runner.cfg, st.project_id) / "current-sidecar" foreign = project_skills_root(runner.cfg, "foreign-project") / "foreign-sidecar" for root, name in ((current, "Current"), (foreign, "Foreign")): root.mkdir(parents=True) (root / "SKILL.md").write_text( f"---\nname: {name}\ndescription: test\n---\n\nRecipe\n", encoding="utf-8", ) (root / "kernel.py").write_text("VALUE = 1\n", encoding="utf-8") policy = runner._kernel_read_isolation(st, include_skill_sidecars=True) assert policy is not None roots = {Path(root).resolve() for root in policy.roots} allowed = {Path(root).resolve() for root in policy.allowed_roots} assert roots == { runner.cfg.data_dir.resolve(), (scratch / "users").resolve(), } assert (scratch / "users" / "alice").resolve() in allowed assert ( kernel_artifact_input_dir(runner.cfg.data_dir, st.root_frame_id).resolve() in allowed ) assert current.resolve() in allowed assert foreign.resolve() not in allowed def test_team_read_policy_rejects_symlinked_personal_namespace_and_root_overlap( monkeypatch, tmp_path, ): canonical_temp = tmp_path / "canonical-temp" canonical_temp.mkdir() monkeypatch.setattr(gateway_mod.tempfile, "gettempdir", lambda: str(canonical_temp)) runner = _runner(tmp_path / "daemon", team_mode=True) st = runner._state("frame-team-symlink", "default") scratch = tmp_path / "scratch" outside = tmp_path / "outside-users" scratch.mkdir() outside.mkdir() (scratch / "users").symlink_to(outside, target_is_directory=True) runner.cfg.data_roots = [scratch] with pytest.raises(RuntimeError, match="personal-data namespace"): runner._kernel_read_isolation(st) runner.cfg.data_roots = [runner.cfg.data_dir / "nested-data-root"] (runner.cfg.data_roots[0]).mkdir() with pytest.raises(RuntimeError, match="must not overlap"): runner._kernel_read_isolation(st) temporary_data_root = canonical_temp / "lab-root" temporary_data_root.mkdir() runner.cfg.data_roots = [temporary_data_root] with pytest.raises(RuntimeError, match="canonical system temporary"): runner._kernel_read_isolation(st) def test_team_read_policy_rejects_symlinked_artifact_input_scope(tmp_path): runner = _runner(tmp_path / "daemon", team_mode=True) st = runner._state("frame-team-artifact-symlink", "default") outside = tmp_path / "outside-artifact-inputs" outside.mkdir() session_inputs = kernel_artifact_input_dir(runner.cfg.data_dir, st.root_frame_id) session_inputs.parent.mkdir() session_inputs.symlink_to(outside, target_is_directory=True) with pytest.raises(RuntimeError, match="Artifact input scope"): runner._kernel_read_isolation(st) def test_r_slot_is_lazy_reused_and_soft_fails_without_touching_python( monkeypatch, tmp_path ): from openai4s.kernel import environments as envmod from openai4s.kernel import r_kernel as r_kernel_mod runner = _runner(tmp_path) st = runner._state("frame-r", "default") st.dispatcher = SimpleNamespace(active_r_env=None) python = _RecordingKernel("python") py_lease = st.kernels.ensure("python", "base", lambda: python) created = [] def get_environment(name): return SimpleNamespace(name=name) if name else None def spawn_r_kernel(*, cwd, env, read_isolation=None): assert read_isolation is None name = env.name if env is not None else "default" if name == "broken": raise RuntimeError("R is missing") kernel = _RecordingKernel(name) created.append(kernel) return kernel monkeypatch.setattr(envmod, "get_environment", get_environment) monkeypatch.setattr(r_kernel_mod, "spawn_r_kernel", spawn_r_kernel) monkeypatch.setattr( gateway_mod, "bootstrap_r_generation", lambda _kernels, _workspace, _lease: {}, ) assert runner._ensure_r_kernel(st) is None first_r = st.r_kernel assert runner._ensure_r_kernel(st) is None assert st.r_kernel is first_r and len(created) == 1 st.dispatcher.active_r_env = "r-special" assert runner._ensure_r_kernel(st) is None second_r = st.r_kernel assert second_r is not first_r assert first_r.shutdown_calls == 1 st.dispatcher.active_r_env = "broken" error = runner._ensure_r_kernel(st) assert error == "R kernel unavailable: R is missing" assert st.r_kernel is second_r and second_r.shutdown_calls == 0 assert st.r_env_name == "r-special" assert st.kernels.lease("python") == py_lease def test_r_execution_exception_shuts_down_the_exact_desynchronized_lease(tmp_path): runner = _runner(tmp_path) st = runner._state("frame-r-error", "default") st.dispatcher = SimpleNamespace(active_r_env=None) class BrokenR(_RecordingKernel): def execute(self, code, origin="agent", on_chunk=None, *, cell_id=None): raise RuntimeError("malformed protocol frame") kernel = BrokenR("r") st.kernels.ensure("r", None, lambda: kernel) runner._ensure_r_kernel = lambda state: None with pytest.raises(RuntimeError, match="malformed protocol"): runner._execute_and_log( st, "stop('bad frame')", "agent", lambda event: None, stream=False, language="r", ) assert st.r_kernel is None assert kernel.shutdown_calls == 1 def test_watchdog_passes_canonical_cell_id_to_kernel(tmp_path): runner = _runner(tmp_path) state = runner._state("frame-cell-id", "default") seen = [] class ImmediateKernel: def is_alive(self): return True def execute(self, code, origin="agent", on_chunk=None, *, cell_id=None): seen.append((code, origin, cell_id)) return {"id": cell_id, "stdout": "", "stderr": "", "error": None} def interrupt(self): pass def shutdown(self): pass state.kernels.ensure("python", "base", ImmediateKernel) result = runner._execute_with_watchdog( state, "print('identified')", "agent", None, cell_id="cell-shared", ) assert result["id"] == "cell-shared" assert seen == [("print('identified')", "agent", "cell-shared")] def test_watchdog_hard_kill_restarts_exact_python_lease(monkeypatch, tmp_path): runner = _runner(tmp_path) st = runner._state("frame-watchdog", "default") class HungKernel(_RecordingKernel): def execute(self, code, origin="agent", on_chunk=None, *, cell_id=None): self.execute_entered.set() assert self.execute_release.wait(2) raise RuntimeError("worker pipe closed") def interrupt(self) -> None: self.interrupt_calls += 1 kernel = HungKernel("python") first = st.kernels.ensure("python", "base", lambda: kernel) bootstrapped = [] monkeypatch.setenv("OPENAI4S_CELL_TIMEOUT", "0.01") monkeypatch.setattr(gateway_mod, "_WATCHDOG_INTERRUPT_GRACE_S", 0.01) monkeypatch.setattr(gateway_mod, "_WATCHDOG_KILL_GRACE_S", 0.1) monkeypatch.setattr( runner, "_run_bootstrap", lambda state, target=None: bootstrapped.append(target) ) with pytest.raises(TimeoutError, match="cell exceeded"): runner._execute_with_watchdog(st, "hang()", "agent", None) recovered = st.kernels.lease("python") assert recovered is not None and recovered.kernel is first.kernel assert recovered.generation == 1 assert kernel.interrupt_calls == kernel.kill_calls == kernel.restart_calls == 1 assert bootstrapped == [kernel]