From 3247dc442231b063426a215d068a50402bb0a9bb Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 19:30:25 +0800 Subject: [PATCH] test: isolate machine CLI smokes from model context transport Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../autonomous-replan-obligation-smoke.py | 10 ++++- ...h-state-shared-runtime-projection-smoke.py | 2 + .../test_canary_cli_transport.py | 37 +++++++++++++++++++ 3 files changed, 47 insertions(+), 2 deletions(-) diff --git a/examples/autonomous-replan-obligation-smoke.py b/examples/autonomous-replan-obligation-smoke.py index 5512834286..f0dbc3bf34 100644 --- a/examples/autonomous-replan-obligation-smoke.py +++ b/examples/autonomous-replan-obligation-smoke.py @@ -6,11 +6,13 @@ import contextlib import io import json +import os import shlex import subprocess import sys import tempfile from pathlib import Path +from unittest.mock import patch REPO_ROOT = Path(__file__).resolve().parents[1] @@ -48,7 +50,8 @@ def run_cli( if not USE_SUBPROCESS_CLI: stdout = io.StringIO() stderr = io.StringIO() - with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): + with patch.dict(os.environ, {"LOOPX_CONTEXT_DELIVERY": "off"}), \ + contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): exit_code = loopx_cli.main(argv) if exit_code != 0: raise AssertionError( @@ -66,6 +69,7 @@ def run_cli( capture_output=True, text=True, timeout=timeout, + env={**os.environ, "LOOPX_CONTEXT_DELIVERY": "off"}, ) except subprocess.TimeoutExpired as exc: raise AssertionError( @@ -84,7 +88,8 @@ def run_cli_error( if not USE_SUBPROCESS_CLI: stdout = io.StringIO() stderr = io.StringIO() - with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): + with patch.dict(os.environ, {"LOOPX_CONTEXT_DELIVERY": "off"}), \ + contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): exit_code = loopx_cli.main(argv) assert exit_code != 0, stdout.getvalue() return "\n".join((stderr.getvalue(), stdout.getvalue())) @@ -96,6 +101,7 @@ def run_cli_error( capture_output=True, text=True, timeout=timeout, + env={**os.environ, "LOOPX_CONTEXT_DELIVERY": "off"}, ) assert result.returncode != 0, result.stdout return "\n".join((result.stderr, result.stdout)) diff --git a/examples/control_plane/refresh-state-shared-runtime-projection-smoke.py b/examples/control_plane/refresh-state-shared-runtime-projection-smoke.py index 71aa6e5abf..a5955697ee 100644 --- a/examples/control_plane/refresh-state-shared-runtime-projection-smoke.py +++ b/examples/control_plane/refresh-state-shared-runtime-projection-smoke.py @@ -49,6 +49,7 @@ def run_cli(*args: str, cwd: Path, shared_runtime: Path) -> dict: **os.environ, "PYTHONPATH": str(REPO_ROOT), "LOOPX_RUNTIME_ROOT": str(shared_runtime), + "LOOPX_CONTEXT_DELIVERY": "off", }, ) assert result.returncode == 0, result.stdout + result.stderr @@ -68,6 +69,7 @@ def run_cli_failure(*args: str, cwd: Path, shared_runtime: Path) -> dict: **os.environ, "PYTHONPATH": str(REPO_ROOT), "LOOPX_RUNTIME_ROOT": str(shared_runtime), + "LOOPX_CONTEXT_DELIVERY": "off", }, ) assert result.returncode != 0, result.stdout + result.stderr diff --git a/tests/control_plane/test_canary_cli_transport.py b/tests/control_plane/test_canary_cli_transport.py index a1da7b10cc..c093e08ccb 100644 --- a/tests/control_plane/test_canary_cli_transport.py +++ b/tests/control_plane/test_canary_cli_transport.py @@ -3,6 +3,9 @@ import hashlib import json +import os +from pathlib import Path +import runpy import subprocess import sys from types import SimpleNamespace @@ -44,3 +47,37 @@ def test_real_capture_cli_returns_the_same_wire_through_the_helper(tmp_path, mon assert observed == json.loads(direct.stdout) assert capture.read_bytes() == raw assert not registry.exists() and not runtime.exists() + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("gate", ["ready", "user_gate", "zero_quota"]) +@pytest.mark.parametrize("parent_mode", ["on", "off"]) +@pytest.mark.parametrize("consumer", ["autonomous", "autonomous_subprocess", "shared_runtime"]) +def test_machine_smoke_calls_keep_core_gates_and_parent_context_mode( + tmp_path, monkeypatch, provider, gate, parent_mode, consumer, +): + from tests.control_plane import test_quota_settlement_cli as fixture + + project, runtime, registry = fixture._write_fixture(tmp_path) + if gate == "user_gate": + fixture._append_blocking_user_gate(project) + elif gate == "zero_quota": + source = json.loads(registry.read_text()) + source["goals"][0]["quota"]["compute"] = 0 + source["goals"][0]["quota"].pop("allowed_slots") + registry.write_text(json.dumps(source)) + canary_harness.import_fixture_authority(registry, fixture.GOAL_ID, provider=provider) + monkeypatch.setenv("LOOPX_CONTEXT_DELIVERY", parent_mode) + root = Path(__file__).resolve().parents[2] + arguments = ("quota", "should-run", "--goal-id", fixture.GOAL_ID, + "--agent-id", fixture.AGENT_ID) + if consumer == "shared_runtime": + smoke = runpy.run_path(str(root / "examples/control_plane/refresh-state-shared-runtime-projection-smoke.py")) + observed = smoke["run_cli"]("--registry", str(registry), "--runtime-root", str(runtime), + *arguments, cwd=project, shared_runtime=runtime) + else: + smoke = runpy.run_path(str(root / "examples/autonomous-replan-obligation-smoke.py")) + smoke["run_cli"].__globals__["USE_SUBPROCESS_CLI"] = consumer == "autonomous_subprocess" + observed = smoke["run_cli"](*arguments, registry_path=registry, runtime=runtime) + assert observed["normal_delivery_allowed"] is (gate == "ready"), observed + assert os.environ["LOOPX_CONTEXT_DELIVERY"] == parent_mode