diff --git a/benchmark/tests/test_checkpoint_joint_receiver.py b/benchmark/tests/test_checkpoint_joint_receiver.py index 8c7295d40e..0bbb16d49a 100644 --- a/benchmark/tests/test_checkpoint_joint_receiver.py +++ b/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. diff --git a/benchmark/tests/test_native_checkpoint_handoff.py b/benchmark/tests/test_native_checkpoint_handoff.py index 26c65ca93a..31f5799577 100644 --- a/benchmark/tests/test_native_checkpoint_handoff.py +++ b/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, diff --git a/scripts/external_scheduler_worker.py b/scripts/external_scheduler_worker.py index 5f6822e25c..9a7e3a125c 100644 --- a/scripts/external_scheduler_worker.py +++ b/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 diff --git a/tests/control_plane/test_context_machine_configuration.py b/tests/control_plane/test_context_machine_configuration.py index 9763d8b6e4..ededa3c5cf 100644 --- a/tests/control_plane/test_context_machine_configuration.py +++ b/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: diff --git a/tests/control_plane/test_scheduler_context_observation.py b/tests/control_plane/test_scheduler_context_observation.py new file mode 100644 index 0000000000..9634e207a1 --- /dev/null +++ b/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