test: isolate machine CLI smokes from model context transport

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))

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

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