fix(scheduler): keep admission observations inline

benchmark/tests/test_checkpoint_joint_receiver.py

@@ -984,14 +984,17 @@ def run_native_work(current_guard, logs, purpose):
                     goal,
                     "--agent-id",
                     agent,
-                    "--turn-instance-id",
-                    "joint-next-audit",
+                    # Readback is a scheduler observation, not another work
+                    # Turn whose unperformed closeout could block recapture.
+                    "--context-delivery",
+                    "off",
                     "--runtime-profile",
                     "generic_cli",
                     "--available-capability",
                     "shell",
                 )
                 assert final["interaction_contract"]["mode"] == "bounded_delivery"
+                assert not final.get("heartbeat_receipt", {}).get("closeout_required")
             # The next genuinely new decision boundary can capture changed
             # research/code/native history, after actual useful continuation.
             # This partial point deliberately makes no evaluator qualification.

benchmark/tests/test_native_checkpoint_handoff.py

@@ -405,7 +405,7 @@ def test_actual_worker_captures_pending_replan_before_native_model(tmp_path, exp
     executable.chmod(0o700)
     skills = tmp_path / "skills"
     skills.mkdir()
-    requests, guards, work_turns = [], [], []
+    requests, guards, work_turns, process_continuations = [], [], [], []
     phase = {"kind": "initial", "values": [], "tool_sent": False}
     reader = NativeContextReader()
 
@@ -428,6 +428,7 @@ def run(*args):
     def response(payload, index, message):
         requests.append(payload)
         if continuation := native_process_continuation(payload, f"checkpoint_{index}"):
+            process_continuations.append(index)
             return continuation
         if context_delivery == "on" and phase["kind"] != "initial":
             follow = reader.follow(payload)
@@ -541,7 +542,7 @@ def settle_work(result, expected):
             settle_work(prior, value)
         assert len(set(work_turns)) == 6
         if context_delivery == "off":
-            assert len(calls) == 13
+            assert len(calls) == 13 + len(process_continuations)
         else:
             assert reader.deliveries[0]["mode"] == "full"
             assert any(row["mode"] == "delta" for row in reader.deliveries[1:])
@@ -578,7 +579,7 @@ def settle_work(result, expected):
         assert result["session"]["session_id"] == prior["session"]["session_id"]
         assert not result.get("checkpoint_handoffs")
         if context_delivery == "off":
-            assert len(calls) == 15
+            assert len(calls) == 15 + len(process_continuations)
         assert cli._heartbeat_receipt_count(runtime, result["turn_id"]) == 1
         assert "settlement_binding_owed" in json.dumps(requests[-1]) and len(guards) == 1
 
@@ -628,7 +629,7 @@ def settle_work(result, expected):
         settle_work(adopted, 8)
         assert len(work_turns) == 7
         if context_delivery == "off":
-            assert len(calls) == 17
+            assert len(calls) == 17 + len(process_continuations)
         assert all(json.loads((project / f"evidence-{v}.json").read_text())["value"] == v for v in (7, 8))
         (tmp_path / "controller-result.json").write_text(json.dumps({**result["checkpoint"],
             "natural_work_turns": work_turns[:6], "stride_action": "grow", "explore_mode": explore_mode,

scripts/external_scheduler_worker.py

@@ -675,6 +675,10 @@ def build_should_run_command(args: argparse.Namespace) -> list[str]:
             args.runtime_profile,
             "--include-detail",
             "scheduler",
+            # This observer consumes the decision itself. The admitted wake
+            # owns its Turn, native context and capture; keep its env unchanged.
+            "--context-delivery",
+            "off",
         ]
     )
     return command

tests/control_plane/test_context_machine_configuration.py

@@ -467,8 +467,10 @@ def test_explicit_on_unknown_host_fails_before_first_use_directory_creation(tmp_
     ("file", None, False, False), ("sqlite", None, False, False),
 ])
 def test_native_worker_default_on_and_explicit_off_isolation(tmp_path, monkeypatch, provider, worker_mode, machine_on, transport_ready):
+    from argparse import Namespace
     from benchmark.runtime.worker import run_once
-    from tests.control_plane.codex_checkpoint_fixture import local_inference_provider
+    from scripts.external_scheduler_worker import build_should_run_command, parse_tick
+    from tests.control_plane.codex_checkpoint_fixture import local_inference_provider, native_process_continuation
     from tests.control_plane.test_codex_context_boundary import _input_text, _native_fixture
     import sys
 
@@ -493,11 +495,24 @@ def test_native_worker_default_on_and_explicit_off_isolation(tmp_path, monkeypat
         def unavailable(*_args):
             raise ValueError("Fixture native hooks unavailable")
         monkeypatch.setattr("benchmark.runtime.context_delivery.prepare_context_delivery", unavailable)
-    observed, errors = [], []
+    # The real outer scheduler observes before the native worker prepares its
+    # session. It inherits the treatment, but must consume an inline decision.
+    observer_env = {**env, "PYTHONPATH": str(Path(__file__).resolve().parents[2]),
+        "LOOPX_CONTEXT_DELIVERY": "off" if worker_mode == "off" or not transport_ready else "on"}
+    observer = build_should_run_command(Namespace(cli_bin=program, registry=str(registry),
+        runtime_root=str(runtime), goal_id=GOAL_ID, agent_id=AGENT_ID, runtime_profile="generic_cli"))
+    observed_guard = subprocess.run(observer, cwd=project, env=observer_env,
+        capture_output=True, text=True, timeout=30, check=False)
+    assert observed_guard.returncode == 0, observed_guard.stderr or observed_guard.stdout
+    assert parse_tick(json.loads(observed_guard.stdout)).should_run
+    observed, errors, process_continuations = [], [], []
     saved = workspace / "original-guard.json"
 
     def response(request, number, message):
         try:
+            if continuation := native_process_continuation(request, f"context-default-{number}"):
+                process_continuations.append(number)
+                return continuation
             if number == 1:
                 body = _input_text(request)
                 command = next(line.strip() for line in body.splitlines() if " quota should-run " in line)
@@ -548,7 +563,7 @@ def response(request, number, message):
         original["OPENAI_BASE_URL"] = url
         result = run_once(env)
     assert result["ok"], result
-    assert not errors and len(requests) == 2, errors
+    assert not errors and len(requests) == 2 + len(process_continuations), errors
     assert env == original, "Resolved worker defaults must not mutate caller/frozen input"
     assert _heartbeat_receipt_count(runtime, observed[0]) == 1 and _spend_run_count(runtime) == 0
     if worker_mode == "off" or not transport_ready:

tests/control_plane/test_scheduler_context_observation.py

@@ -0,0 +1,78 @@
+"""A shell scheduler observes admission; the admitted native wake owns context."""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+
+import pytest
+
+from scripts import external_scheduler_worker as worker
+from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime
+from tests.control_plane.test_quota_settlement_cli import (
+    AGENT_ID, GOAL_ID, TODO_ID, _heartbeat_receipt_count, _spend_run_count, _write_fixture,
+)
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("native_context", ["on", "off"])
+@pytest.mark.skipif(os.name != "posix", reason="Exercises the POSIX shell scheduler and private native files")
+def test_scheduler_observation_preserves_native_wake_context(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, provider: str, native_context: str,
+) -> None:
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    monkeypatch.setenv("LOOPX_CONTEXT_DELIVERY", native_context)
+    monkeypatch.setenv("LOOPX_EXECUTION_MODE", "heartbeat")
+    monkeypatch.setenv("CODEX_HOME", str(tmp_path / "native-home"))
+    repository = Path(__file__).resolve().parents[2]
+    monkeypatch.setenv("PYTHONPATH", str(repository))
+    project, runtime, registry = _write_fixture(tmp_path, provider=provider)
+    cli = tmp_path / "loopx-cli"
+    cli.write_text(f"#!{sys.executable}\nfrom loopx.entrypoint import main\nmain()\n")
+    cli.chmod(0o700)
+    args = argparse.Namespace(cli_bin=str(cli), registry=str(registry),
+        runtime_root=str(runtime), goal_id=GOAL_ID, agent_id=AGENT_ID,
+        runtime_profile="generic_cli")
+
+    # This process has no native Turn/session. Its current decision must remain
+    # directly consumable by the scheduler even when its future wake uses on.
+    original_environment = dict(os.environ)
+    decision = worker.run_quota_should_run(worker.build_should_run_command(args),
+        timeout_seconds=30)
+    assert decision["ok"] and worker.parse_tick(decision).should_run
+    assert decision["selected_todo"]["todo_id"] == TODO_ID
+    assert not (runtime / "context-delivery").exists()
+    assert dict(os.environ) == original_environment
+    assert _spend_run_count(runtime) == 0
+
+    # The native wake still captures its own exact Turn. The observer's off
+    # declaration must never become a global opt-out or mutate the child env.
+    turn = "native-wake-after-observation"
+    command = [str(cli), "--format", "json", "--registry", str(registry),
+        "--runtime-root", str(runtime), "quota", "should-run", "--goal-id", GOAL_ID,
+        "--agent-id", AGENT_ID, "--runtime-profile", "generic_cli",
+        "--turn-instance-id", turn, "--context-delivery", native_context]
+    if native_context == "on":
+        from loopx.control_plane.turn_driver.codex_context import codex_context_hook
+
+        (tmp_path / "captures").mkdir(mode=0o700)
+        (tmp_path / "context").mkdir(mode=0o700)
+        codex_context_hook(str(tmp_path / "context"), {"hook_event_name": "SessionStart",
+            "source": "startup", "session_id": "native-context-wake", "cwd": str(project)})
+        command += ["--decision-output-root", str(tmp_path / "captures"),
+            "--codex-context-dir", str(tmp_path / "context")]
+    result = subprocess.run(command, cwd=project, capture_output=True, text=True,
+        env=os.environ.copy(), timeout=30, check=False)
+    assert result.returncode == 0, result.stderr or result.stdout
+    payload = json.loads(result.stdout)
+    assert payload["ok"], json.dumps({k: payload.get(k) for k in ("reason", "error_code", "state")})
+    if native_context == "on":
+        assert payload["schema_version"] == "context_capture_navigation_v1"
+        assert Path(payload["reference_file"]).is_file()
+    else:
+        assert payload["selected_todo"]["todo_id"] == TODO_ID
+    assert _heartbeat_receipt_count(runtime, turn) == 1
+    assert _spend_run_count(runtime) == 0