diff --git a/benchmark/README.md b/benchmark/README.md index 45cdd1dbdb..c76b079dbe 100644 --- a/benchmark/README.md +++ b/benchmark/README.md @@ -1038,6 +1038,31 @@ Fresh credentials come from the ordinary runner credential source. The native activation's absolute deadline includes receiver installation and preparation; restarting the entry cannot grant another full budget or submission allowance. +To continue with a different model or reasoning effort, retain the exact sealed +source settings in `--restore-config` and explicitly pass +`--restore-model-config DESTINATION_CONFIG`. Only the top-level `model` and +`model_reasoning_effort` may differ; provider, tools, permissions and other +settings must remain semantically identical. The destination model/effort must +also match `--model`/`--effort`. The native session participant records both +configuration digests, verifies the original session profile and full retained +history, then binds the new profile to a distinct destination session. Retry +uses that same source/destination intent; it cannot switch an already running +receiver. Without the option, exact-configuration admission is unchanged. +The host controller exposes the same explicit transition as +`--destination-model-config`; this remains a CLI/Host journey, with frontend +restore support still open. Changing the model changes the experiment treatment: +do not claim a matched mechanism comparison solely from shared checkpoint data. +It grants no extra native submissions, cooldown relief, permission or time. + +切换模型或思考深度时，`--restore-config` 仍传封存源配置，另显式传入 +`--restore-model-config DESTINATION_CONFIG`。仅顶层 `model` 与 +`model_reasoning_effort` 可变化，其余 provider、工具、权限及配置语义保持一致， +目标选择还须匹配 `--model`/`--effort`。接收端保留源配置摘要和完整历史校验， +在全新会话绑定目标配置；恢复重试不得换意图或改写已运行接收端。省略选项仍要求 +配置完全一致，不增加评测机会、豁免冷却、权限或预算。宿主 CLI 对应选项为 +`--destination-model-config`；frontend restore 旅程仍未交付。模型变化是新的实验因素， +共享 checkpoint 本身不能证明机制收益。 + Warm verification hashes the complete checkpoint with a read window of up to 60 seconds, capped by the original execution deadline. It rejects a read that finishes after that deadline. The launcher reuses this verified metadata before diff --git a/benchmark/edgebench/run.py b/benchmark/edgebench/run.py index eb692b9d19..dc319f30e1 100644 --- a/benchmark/edgebench/run.py +++ b/benchmark/edgebench/run.py @@ -161,6 +161,8 @@ def main(argv=None): retention.add_argument("--no-checkpoint-retention", action="store_true", help="Keep every captured point; disable default checkpoint rotation") parser.add_argument("--restore-config", type=Path) + parser.add_argument("--restore-model-config", type=Path, + help="Explicit destination config for changing only model/effort in a new checkpoint receiver; --restore-config retains the sealed source config") parser.add_argument("--restore-goal") parser.add_argument("--restore-agent") parser.add_argument("--restore-source-todo") @@ -232,6 +234,8 @@ def main(argv=None): parser.error("Research component restoration requires an explicit warm checkpoint continuation") if args.restore_fresh_replan and not args.restore_checkpoint: parser.error("Fresh context requires an explicit warm checkpoint continuation") + if args.restore_model_config is not None and not args.restore_checkpoint: + parser.error("Model transition requires an explicit warm checkpoint continuation") if args.restore_checkpoint and (not handoff or args.eval_interval != 0 or args.replan_after_todos is not None): parser.error("Warm continuation requires full native checkpoint capture, no periodic evaluation and effective-Turn cadence") # One directory is one attempt: never reuse native registration or overwrite @@ -311,8 +315,16 @@ def main(argv=None): candidate = WarmStart(args.restore_checkpoint, args.restore_sha256, args.restore_config, args.restore_goal, args.restore_agent, args.restore_source_todo, int(time.time()) + args.timeout, args.replan_after_turns or 6, - tuple(args.restore_research_component), args.restore_fresh_replan) + tuple(args.restore_research_component), args.restore_fresh_replan, + destination_config=args.restore_model_config) metadata = candidate.verify(task.cwd) + if args.restore_model_config is not None: + from loopx.hosts.codex_checkpoint import checkpoint_destination_config + _, transition = checkpoint_destination_config(args.restore_config, + metadata["native_session"]["config_sha256"], args.restore_model_config) + if (transition["destination_model"] != args.model or + transition["destination_effort"] != args.effort): + raise ValueError("Checkpoint destination model must match the explicit runner model/effort") snapshot = metadata["worker_file_snapshot"] capsule = args.restore_checkpoint / "components" / snapshot["native_capsule_component_id"] clock = attempt / "restored-phase-deadline" diff --git a/benchmark/runtime/sforge_warm_start.py b/benchmark/runtime/sforge_warm_start.py index 04bafb9463..ecccaa0f30 100644 --- a/benchmark/runtime/sforge_warm_start.py +++ b/benchmark/runtime/sforge_warm_start.py @@ -45,6 +45,7 @@ class WarmStart: normal_cadence: int = 6 research_component_ids: tuple[str, ...] = () fresh_replan: bool = False + destination_config: Path | None = None def verify(self, workspace: str) -> dict: from loopx.hosts.checkpoint_retention import checkpoint_bank_use @@ -60,8 +61,10 @@ def _verify(self, workspace: str) -> dict: or any(not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}", v) for v in (self.goal_id, self.agent_id, self.source_todo_id))): raise ValueError("Warm start needs fresh explicit identities and its original open deadline") - if any(not path.is_absolute() or path.resolve(strict=True) != path - for path in (self.root, self.config)): + paths = (self.root, self.config) + if self.destination_config is not None: + paths += (self.destination_config,) + if any(not path.is_absolute() or path.resolve(strict=True) != path for path in paths): raise ValueError("Warm start requires physical absolute participant paths") # Hashing a complete checkpoint can exceed the ordinary RPC budget. # This read allowance never extends the original execution deadline. @@ -74,9 +77,10 @@ def _verify(self, workspace: str) -> dict: metadata = json.loads((self.root / "checkpoint.json").read_bytes())["point_metadata"] native, source = metadata["native_session"], metadata["loopx_state"] if (native["guest_workspace"] != workspace or native["guest_home"] != _CODEX_HOME - or source["goal_id"] == self.goal_id or source["agent_id"] == self.agent_id - or hashlib.sha256(self.config.read_bytes()).hexdigest() != native["config_sha256"]): + or source["goal_id"] == self.goal_id or source["agent_id"] == self.agent_id): raise ValueError("Warm start paths, configuration or destination lineage differ") + from loopx.hosts.codex_checkpoint import checkpoint_destination_config + checkpoint_destination_config(self.config, native["config_sha256"], self.destination_config) _workspace_component(self.root, metadata, Path(workspace)) _research_components(self.root, metadata, Path(workspace), self.research_component_ids) context = metadata["research_context"] @@ -110,6 +114,12 @@ async def _stage(self, environment, workspace: str) -> list[str]: goal_id=self.goal_id, agent_id=self.agent_id, source_todo_id=self.source_todo_id, deadline_epoch=self.deadline_epoch, normal_cadence=self.normal_cadence) + protected_files = [config, f"{_CONTROL}/warm-start.json"] + if self.destination_config is not None: + target = f"{_CONTROL}/restored-model-config.toml" + await environment.upload_file(self.destination_config, target) + payload["destination_config"] = target + protected_files.append(target) if self.research_component_ids: payload["research_component_ids"] = list(self.research_component_ids) if self.fresh_replan: @@ -118,12 +128,12 @@ async def _stage(self, environment, workspace: str) -> list[str]: path = Path(folder) / "input.json" path.write_text(json.dumps(payload)) await environment.upload_file(path, f"{_CONTROL}/warm-start.json") + protected = shlex.join(protected_files) result = await environment.exec( # Modes are part of the sealed digest. Change transport ownership, # never widen modes (native stores commonly contain 0600 members). f"chown -R -P agent:agent {shlex.quote(unit)} && " - f"chown agent:agent {config} {_CONTROL}/warm-start.json && " - f"chmod 0400 {config} {_CONTROL}/warm-start.json", user="root") + f"chown agent:agent {protected} && chmod 0400 {protected}", user="root") if result.return_code: raise RuntimeError("Could not stage the verified checkpoint participants") # The root-only materializer preserves sealed modes. Its receiver must @@ -290,7 +300,9 @@ def admit_warm_session(env: dict[str, str], declaration: dict) -> dict: timeout = _phase_read_timeout(declaration, 90) result = admit_native_checkpoint_session(env, Path(declaration["root"]), declaration["sha256"], - Path(declaration["config"]), sandbox=env["LOOPX_CODEX_SANDBOX"], timeout=timeout) + Path(declaration["config"]), sandbox=env["LOOPX_CODEX_SANDBOX"], timeout=timeout, + **({"destination_config": Path(declaration["destination_config"])} + if declaration.get("destination_config") is not None else {})) _phase_read_timeout(declaration, 90) return result diff --git a/benchmark/tests/test_checkpoint_binding_lock.py b/benchmark/tests/test_checkpoint_binding_lock.py index 014ccb2042..421f756481 100644 --- a/benchmark/tests/test_checkpoint_binding_lock.py +++ b/benchmark/tests/test_checkpoint_binding_lock.py @@ -9,8 +9,9 @@ @pytest.mark.parametrize("replay", [False, True]) @pytest.mark.parametrize("change_goal", [False, True]) +@pytest.mark.parametrize("model_transition", [False, True]) def test_checkpoint_binding_and_readback_recheck_destination( - tmp_path, monkeypatch, capsys, replay, change_goal, + tmp_path, monkeypatch, capsys, replay, change_goal, model_transition, ): from loopx.cli import main from loopx.control_plane import effect_runtime @@ -25,7 +26,9 @@ def test_checkpoint_binding_and_readback_recheck_destination( for path in (workspace, home, unit): path.mkdir() config = home / "config.toml" - config.write_text("[features]\ngoals=false\n") + config.write_text('model="synthetic"\nmodel_reasoning_effort="high"\n[features]\ngoals=false\n') + source_config = tmp_path / "source.toml" + source_config.write_bytes(config.read_bytes()) prefix = ["--format", "json", "--registry", str(registry), "--runtime-root", str(runtime)] def cli(*args): @@ -53,11 +56,24 @@ def cli(*args): (unit / "checkpoint.json").write_text(json.dumps({"point_metadata": { "session": {"session_profile_digest": prepared.digest}, "native_session": native, "loopx_state": {"goal_id": "source", "agent_id": "source-agent"}}})) + destination = None + if model_transition: + destination = tmp_path / "destination.toml" + destination.write_text(source_config.read_text().replace('"synthetic"', '"synthetic-next"') + .replace('"high"', '"max"')) + config.write_bytes(destination.read_bytes()) + env.update(MODEL_NAME="synthetic-next", REASONING_EFFORT="max") + destination_profile = session_type(env, context="resume", sandbox="read-only", receipt={}).digest + transition = None + if model_transition: + transition = {**codex_checkpoint.checkpoint_destination_config( + source_config, native["config_sha256"], destination)[1], + "source_profile_digest": prepared.digest, "destination_profile_digest": destination_profile} if replay: _store_codex_cli_session(runtime, lineage=prepared.lineage, session_scope="agent", session_id="destination-session", goal_ref=prepared.goal_ref, - session_profile_digest=prepared.digest, checkpoint_artifact_sha256=digest, - checkpoint_artifact_root=str(unit)) + session_profile_digest=destination_profile, checkpoint_artifact_sha256=digest, + checkpoint_artifact_root=str(unit), checkpoint_model_transition=transition) before = select_codex_cli_session(runtime, lineage=prepared.lineage, session_scope="agent") # Native storage/inference is outside this lock regression. Registry, @@ -74,6 +90,7 @@ def verify(name, payload, **kwargs): def fork(*args, **kwargs): forks.append(True) + assert kwargs.get("destination_config") == destination return {"session_id": "destination-session", "model_calls": 0} def construct(*args, **kwargs): @@ -85,18 +102,26 @@ def construct(*args, **kwargs): monkeypatch.setattr(effect_runtime, "effect_runtime_result", verify) monkeypatch.setattr(codex_checkpoint, "fork_native_checkpoint", fork) - monkeypatch.setattr(codex_checkpoint, "native_session_checkpoint", lambda *a, **k: {"metadata": native}) + monkeypatch.setattr(codex_checkpoint, "native_session_checkpoint", lambda *a, **k: { + "metadata": {**native, "config_sha256": hashlib.sha256(config.read_bytes()).hexdigest()}}) monkeypatch.setattr(codex_agent_session, "CodexAgentSession", construct) if change_goal: with pytest.raises(ValueError, match="Goal configuration changed"): codex_checkpoint.admit_native_checkpoint_session( - env, unit, digest, config, sandbox="read-only", timeout=30) + env, unit, digest, source_config, sandbox="read-only", timeout=30, + **({"destination_config": destination} if destination is not None else {})) assert select_codex_cli_session(runtime, lineage=prepared.lineage, session_scope="agent") == before else: result = codex_checkpoint.admit_native_checkpoint_session( - env, unit, digest, config, sandbox="read-only", timeout=30) + env, unit, digest, source_config, sandbox="read-only", timeout=30, + **({"destination_config": destination} if destination is not None else {})) assert result["session_binding_admitted"] and result["session_id"] == "destination-session" binding = select_codex_cli_session(runtime, lineage=prepared.lineage, session_scope="agent") assert binding["checkpoint_artifact_sha256"] == digest assert binding["checkpoint_artifact_root"] == str(unit) + assert binding["session_profile_digest"] == destination_profile + if replay and model_transition: + assert result["model_transition"]["source_model"] == "synthetic" + assert result["model_transition"]["destination_effort"] == "max" + assert (source_config.read_bytes() != config.read_bytes()) is model_transition assert len(forks) == (0 if replay else 1) diff --git a/benchmark/tests/test_checkpoint_model_transition.py b/benchmark/tests/test_checkpoint_model_transition.py new file mode 100644 index 0000000000..e85ef5585f --- /dev/null +++ b/benchmark/tests/test_checkpoint_model_transition.py @@ -0,0 +1,89 @@ +"""A new receiver can choose a model without rewriting its captured evidence.""" +import hashlib + +import pytest + +from loopx.hosts.codex_checkpoint import checkpoint_destination_config + + +@pytest.fixture +def settings(tmp_path): + source, target = tmp_path / "source.toml", tmp_path / "target.toml" + source.write_text('model="source-model"\nmodel_reasoning_effort="high"\n' + 'sandbox_mode="read-only"\nmodel_provider="openai"\n' + '[features]\ngoals=false\n') + target.write_text(source.read_text().replace('"source-model"', '"next-model"') + .replace('"high"', '"max"')) + return source, target, hashlib.sha256(source.read_bytes()).hexdigest() + + +def test_explicit_model_transition_keeps_source_and_all_other_settings(settings): + source, target, digest = settings + before = source.read_bytes() + data, transition = checkpoint_destination_config(source, digest, target) + assert data == target.read_bytes() and source.read_bytes() == before + assert transition == { + "source_config_sha256": digest, + "destination_config_sha256": hashlib.sha256(data).hexdigest(), + "source_model": "source-model", "source_effort": "high", + "destination_model": "next-model", "destination_effort": "max", + } + assert checkpoint_destination_config(source, digest) == (before, None) + + +@pytest.mark.parametrize("old,new", [ + ('"read-only"', '"danger-full-access"'), ('"openai"', '"other"'), + ('goals=false', 'goals=true'), +]) +def test_model_transition_does_not_change_permissions_provider_or_native_goal(settings, old, new): + source, target, digest = settings + target.write_text(target.read_text().replace(old, new)) + with pytest.raises(ValueError, match="only model and reasoning effort"): + checkpoint_destination_config(source, digest, target) + + +def test_model_transition_preserves_source_digest_and_rejects_redirected_target(settings, tmp_path): + source, target, digest = settings + with pytest.raises(ValueError, match="effective configuration differs"): + checkpoint_destination_config(target, digest, source) + link = tmp_path / "target-link.toml" + link.symlink_to(target) + with pytest.raises(ValueError, match="physical absolute path"): + checkpoint_destination_config(source, digest, link) + + +@pytest.mark.parametrize("replacement", ['model=""', 'model=true', 'model=13']) +def test_model_transition_requires_an_explicit_model(settings, replacement): + source, target, digest = settings + target.write_text(target.read_text().replace('model="next-model"', replacement)) + with pytest.raises(ValueError, match="explicit model and reasoning effort"): + checkpoint_destination_config(source, digest, target) + + +def test_transition_provenance_survives_exact_session_observations_and_qualified_refresh(tmp_path): + from loopx.control_plane.turn_driver.codex_sessions import ( + _replace_checkpoint_codex_session, _store_codex_cli_session, load_codex_cli_session, + ) + facts = {"source_config_sha256": "a" * 64, "destination_config_sha256": "b" * 64, + "source_profile_digest": "c" * 64, "destination_profile_digest": "d" * 64, + "source_model": "original", "source_effort": "high", + "destination_model": "next", "destination_effort": "max"} + runtime, point = tmp_path / "runtime", tmp_path / "point" + point.mkdir() + binding = dict(lineage={"goal_id": "goal", "agent_id": "receiver"}, + session_scope="agent", session_id="fork", session_profile_digest="d" * 64) + _store_codex_cli_session(runtime, **binding, checkpoint_artifact_sha256="e" * 64, + checkpoint_artifact_root=str(point), checkpoint_model_transition=facts) + _store_codex_cli_session(runtime, **binding) + current = load_codex_cli_session(runtime, lineage=binding["lineage"], session_scope="agent") + assert current["checkpoint_model_transition"] == facts + with pytest.raises(ValueError, match="transition cannot change"): + _store_codex_cli_session(runtime, **binding, + checkpoint_model_transition={**facts, "destination_model": "another"}) + _replace_checkpoint_codex_session(runtime, lineage=binding["lineage"], expected=current, + session_id="fresh-context", refresh={"checkpoint_sha256": "e" * 64}) + refreshed = load_codex_cli_session(runtime, lineage=binding["lineage"], session_scope="agent") + assert refreshed["checkpoint_model_transition"] == facts + _store_codex_cli_session(runtime, **{**binding, "session_id": "unrelated"}) + assert "checkpoint_model_transition" not in load_codex_cli_session( + runtime, lineage=binding["lineage"], session_scope="agent") diff --git a/benchmark/tests/test_native_checkpoint.py b/benchmark/tests/test_native_checkpoint.py index cf378f3532..b93d3ca24e 100644 --- a/benchmark/tests/test_native_checkpoint.py +++ b/benchmark/tests/test_native_checkpoint.py @@ -291,7 +291,8 @@ def test_real_native_workspace_readback_rejects_foreign_code_basis(tmp_path): @pytest.mark.skipif(not os.environ.get("LOOPX_NATIVE_CHECKPOINT_BIN"), reason="Requires an explicit real native Host binary; only a local static provider is used") @pytest.mark.parametrize("case", ["resume", "scope_change", "profile_change", "joint", "binding_reply_loss", - "fork_reply_loss", "fork_missing", "fork_ambiguous", "warm_preparation"]) + "fork_reply_loss", "fork_missing", "fork_ambiguous", "warm_preparation", + "model_transition"]) def test_real_fork_admission_and_native_resume_use_retained_state(tmp_path, capsys, monkeypatch, case, request): """Real Host/CLI/Goal/session/artifact owners; deterministic synthetic work. @@ -417,7 +418,7 @@ def response(request, index, message): binary = os.environ["LOOPX_NATIVE_CHECKPOINT_BIN"] # The warm receiver uses the benchmark's execution profile inside its # isolated worker. Other cases retain their original sandbox contract. - sandbox = ("danger-full-access" if case == "warm_preparation" else + sandbox = ("danger-full-access" if case in {"warm_preparation", "model_transition"} else "workspace-write" if case == "joint" else "read-only") execution = Execution(context="resume", sandbox=sandbox) with real_native_source(tmp_path / "native", binary, workspace, close_source=True, @@ -516,6 +517,14 @@ def retained_source(): env.update(LOOPX_REGISTRY=str(registry), LOOPX_RUNTIME_ROOT=str(runtime), LOOPX_GOAL_ID=destination_goal, LOOPX_AGENT_ID=destination_agent, MODEL_NAME="fixture", REASONING_EFFORT="high") + admission_options, model_config_args = {}, [] + if case == "model_transition": + target_config = tmp_path / "destination-model.toml" + target_config.write_text(config.read_text().replace('model="fixture"', 'model="fixture-next"') + .replace('model_reasoning_effort="high"', 'model_reasoning_effort="max"')) + env.update(MODEL_NAME="fixture-next", REASONING_EFFORT="max") + admission_options = {"destination_config": target_config} + model_config_args = ["--destination-model-config", str(target_config)] lifecycle_registry = tmp_path / "lifecycle-only" / "registry.json" assert main(["--format", "json", "--registry", str(lifecycle_registry), "--runtime-root", str(tmp_path / "lifecycle-only/runtime"), "project", "register", @@ -619,7 +628,8 @@ def lose_binding_reply(self, *args): "agent_id": destination_agent}, session_scope="agent")["checkpoint_artifact_sha256"] == digest assert len(calls) == 1 admitted_cli = subprocess.run([sys.executable, "-m", "loopx.hosts.codex_checkpoint", - "--root", str(unit), "--expected-sha256", digest, "--config-path", str(config), "--sandbox", execution.sandbox], + "--root", str(unit), "--expected-sha256", digest, "--config-path", str(config), + "--sandbox", execution.sandbox, *model_config_args], env=env, text=True, capture_output=True, timeout=40) assert admitted_cli.returncode == 0, admitted_cli.stderr admitted = json.loads(admitted_cli.stdout) @@ -636,7 +646,8 @@ def lose_binding_reply(self, *args): def forbidden_fork(*args, **kwargs): raise AssertionError("a repeated admission must not create a second native fork") replay.setattr("loopx.hosts.codex_checkpoint.fork_native_checkpoint", forbidden_fork) - recovered = admit_native_checkpoint_session(env, unit, digest, config, sandbox=(execution).sandbox, timeout=30) + recovered = admit_native_checkpoint_session(env, unit, digest, config, + sandbox=(execution).sandbox, timeout=30, **admission_options) assert recovered == admitted assert bound["checkpoint_artifact_sha256"] == digest # Transport reply loss leaves the atomic source-bound binding as the @@ -644,9 +655,9 @@ def forbidden_fork(*args, **kwargs): retry_cli = subprocess.run(admitted_cli.args, env=env, text=True, capture_output=True, timeout=40) assert retry_cli.returncode == 0, retry_cli.stderr assert json.loads(retry_cli.stdout) == admitted - with pytest.raises(ValueError, match="profile changed"): + with pytest.raises(ValueError, match="profile changed|destination model differs"): admit_native_checkpoint_session({**env, "REASONING_EFFORT": "low"}, unit, digest, - config, sandbox=(execution).sandbox, timeout=30) + config, sandbox=(execution).sandbox, timeout=30, **admission_options) assert len(calls) == 1 and load_codex_cli_session(runtime, lineage={ "goal_id": destination_goal, "agent_id": destination_agent}, session_scope="agent") == bound if case == "resume": @@ -788,6 +799,9 @@ def forbidden_fork(*args, **kwargs): (logs / "stdout.jsonl").write_text(result.stdout) wake.observe(logs / "stdout.jsonl") assert wake.session_id == admitted["session_id"] and len(calls) == 3 + if case == "model_transition": + assert requests[1]["model"] == "fixture-next" + assert requests[1]["reasoning"]["effort"] == "max" retained_message = ("previous Host deadline has passed" if case == "warm_preparation" else "Synthetic source result.") assert retained_message in json.dumps(requests[1]) @@ -842,7 +856,8 @@ def forbidden_fork(*args, **kwargs): "agent_id": destination_agent}, session_scope="agent") assert after_work["checkpoint_artifact_sha256"] == digest with pytest.raises(ValueError, match="advanced or differs"): - admit_native_checkpoint_session(env, unit, digest, config, sandbox=(execution).sandbox, timeout=30) + admit_native_checkpoint_session(env, unit, digest, config, + sandbox=(execution).sandbox, timeout=30, **admission_options) assert load_codex_cli_session(runtime, lineage={"goal_id": destination_goal, "agent_id": destination_agent}, session_scope="agent") == after_work if case != "warm_preparation": diff --git a/benchmark/tests/test_sforge_warm_start.py b/benchmark/tests/test_sforge_warm_start.py index 5b621870cb..6300e4cde5 100644 --- a/benchmark/tests/test_sforge_warm_start.py +++ b/benchmark/tests/test_sforge_warm_start.py @@ -74,7 +74,7 @@ def point(tmp_path): (workspace / "candidate").write_text("retained candidate") (workspace / "candidate").chmod(0o600) config = tmp_path / "config.toml" - config.write_text("[features]\ngoals = false\n") + config.write_text('model="original"\nmodel_reasoning_effort="high"\n[features]\ngoals = false\n') context = tmp_path / "context.json" context.write_text(json.dumps({"snapshot": {"todos": [{"todo_id": "source-todo"}]}})) components = [{"id": "workspace", "source_path": str(workspace)}, @@ -174,8 +174,14 @@ def test_restore_replaces_image_files_preserves_modes_and_refuses_second_attempt assert workspace.with_name("workspace.before-checkpoint").is_dir() -def test_staging_preserves_sealed_modes(point, monkeypatch, tmp_path): +@pytest.mark.parametrize("model_transition", [False, True]) +def test_staging_preserves_sealed_modes(point, monkeypatch, tmp_path, model_transition): start, workspace = point + if model_transition: + destination = tmp_path / "destination.toml" + destination.write_text(start.config.read_text().replace('"original"', '"next"') + .replace('"high"', '"max"')) + start = replace(start, destination_config=destination) control = tmp_path / "control" control.mkdir() monkeypatch.setattr(warm, "_CONTROL", str(control)) @@ -194,6 +200,12 @@ async def exec(self, command, **kwargs): "expected_sha256": start.sha256}) declaration = json.loads((control / "warm-start.json").read_bytes()) assert declaration["deadline_epoch"] == start.deadline_epoch + if model_transition: + assert (control / "restored-model-config.toml").read_bytes() == destination.read_bytes() + assert declaration["destination_config"] == str(control / "restored-model-config.toml") + assert (control / "restored-config.toml").read_bytes() == start.config.read_bytes() + else: + assert "destination_config" not in declaration def test_normal_cadence_requires_accepted_current_replan_and_is_applied_once(tmp_path, monkeypatch): diff --git a/loopx/control_plane/turn_driver/codex_agent_session.py b/loopx/control_plane/turn_driver/codex_agent_session.py index 07680bf078..a88091595c 100644 --- a/loopx/control_plane/turn_driver/codex_agent_session.py +++ b/loopx/control_plane/turn_driver/codex_agent_session.py @@ -178,8 +178,9 @@ def prepare_checkpoint_refresh(self, review_basis: dict | None, attempt: Path, s self.session_id = None self.receipt["session"].update(action="start_new", session_id=None) - def _bind_verified_fork(self, session_id: str, source_profile_digest: str, - checkpoint_artifact_sha256: str, checkpoint_artifact_root: str) -> None: + def _bind_verified_fork(self, session_id: str, verified_profile_digest: str, + checkpoint_artifact_sha256: str, checkpoint_artifact_root: str, + *, model_transition: dict[str, str] | None = None) -> None: """Bind a controller-verified native fork through the existing session owner. This admits the conversation binding only. It does not claim work, @@ -188,7 +189,7 @@ def _bind_verified_fork(self, session_id: str, source_profile_digest: str, """ if self.session_id is not None: raise ValueError("checkpoint session adoption requires an absent destination binding") - require_codex_session_profile({"session_profile_digest": source_profile_digest}, self.digest) + require_codex_session_profile({"session_profile_digest": verified_profile_digest}, self.digest) self.admission.accept_result( lambda: _store_codex_cli_session( self.root, @@ -200,6 +201,7 @@ def _bind_verified_fork(self, session_id: str, source_profile_digest: str, require_absent=True, checkpoint_artifact_sha256=checkpoint_artifact_sha256, checkpoint_artifact_root=checkpoint_artifact_root, + checkpoint_model_transition=model_transition, ) ) self.session_id = session_id diff --git a/loopx/control_plane/turn_driver/codex_sessions.py b/loopx/control_plane/turn_driver/codex_sessions.py index e9cbcbc689..725b4d9526 100644 --- a/loopx/control_plane/turn_driver/codex_sessions.py +++ b/loopx/control_plane/turn_driver/codex_sessions.py @@ -198,6 +198,7 @@ def _store_codex_cli_session( require_absent: bool = False, checkpoint_artifact_sha256: str | None = None, checkpoint_artifact_root: str | None = None, + checkpoint_model_transition: Mapping[str, str] | None = None, ) -> None: if checkpoint_artifact_sha256 is not None and not BARE_SHA256_PATTERN.fullmatch(checkpoint_artifact_sha256): raise ValueError("checkpoint session source digest is invalid") @@ -215,12 +216,14 @@ def _store_codex_cli_session( retained_source = None retained_root = None retained_refresh = None + retained_transition = None if prior and (prior.get("session_id"), prior.get("goal_ref"), prior.get("session_profile_digest")) == ( session_id, dict(goal_ref) if goal_ref is not None else None, session_profile_digest ): retained_source = prior.get("checkpoint_artifact_sha256") retained_root = prior.get("checkpoint_artifact_root") retained_refresh = prior.get("checkpoint_context_refresh") + retained_transition = prior.get("checkpoint_model_transition") if retained_source is not None: if not isinstance(retained_source, str) or not BARE_SHA256_PATTERN.fullmatch(retained_source): raise ValueError("stored checkpoint session source digest is invalid") @@ -231,6 +234,9 @@ def _store_codex_cli_session( raise ValueError("stored checkpoint session source root is invalid") if retained_root is not None and checkpoint_artifact_root not in (None, retained_root): raise ValueError("checkpoint session source root cannot change") + if retained_transition is not None and checkpoint_model_transition not in (None, retained_transition): + raise ValueError("checkpoint model transition cannot change") + transition = checkpoint_model_transition if checkpoint_model_transition is not None else retained_transition _write_codex_cli_session( runtime_root, lineage=lineage, @@ -245,6 +251,7 @@ def _store_codex_cli_session( checkpoint_artifact_sha256=checkpoint_artifact_sha256 or retained_source, checkpoint_artifact_root=checkpoint_artifact_root or retained_root, checkpoint_context_refresh=retained_refresh, + checkpoint_model_transition=transition, ) @@ -270,7 +277,8 @@ def _replace_checkpoint_codex_session( session_profile_digest=current["session_profile_digest"], checkpoint_artifact_sha256=current["checkpoint_artifact_sha256"], checkpoint_artifact_root=current["checkpoint_artifact_root"], - checkpoint_context_refresh={**refresh, "session_id": session_id}) + checkpoint_context_refresh={**refresh, "session_id": session_id}, + checkpoint_model_transition=current.get("checkpoint_model_transition")) def _write_codex_cli_session( @@ -288,8 +296,14 @@ def _write_codex_cli_session( checkpoint_artifact_sha256: str | None = None, checkpoint_artifact_root: str | None = None, checkpoint_context_refresh: dict[str, Any] | None = None, + checkpoint_model_transition: Mapping[str, str] | None = None, ) -> None: schema = _host_session_schema(host) + if checkpoint_model_transition is not None: + validate_checkpoint_model_transition(checkpoint_model_transition) + if (not checkpoint_artifact_sha256 or + checkpoint_model_transition["destination_profile_digest"] != session_profile_digest): + raise ValueError("checkpoint model transition requires its source-bound destination profile") normalized_session_id = _valid_session_id(session_id) if not normalized_session_id: raise ValueError("Codex CLI returned an invalid session id") @@ -323,6 +337,8 @@ def _write_codex_cli_session( payload["checkpoint_artifact_root"] = checkpoint_artifact_root if checkpoint_context_refresh is not None: payload["checkpoint_context_refresh"] = checkpoint_context_refresh + if checkpoint_model_transition is not None: + payload["checkpoint_model_transition"] = dict(checkpoint_model_transition) if operation_profile_digest is not None: payload["operation_transport"] = ( "dsh-native-operation-tools-v0" if host == "dsh" @@ -410,6 +426,17 @@ def read() -> dict[str, Any] | None: return dict(value) +def validate_checkpoint_model_transition(value: Mapping[str, str]) -> None: + """Validate the source/destination facts retained by checkpoint admission.""" + digests = ("source_config_sha256", "destination_config_sha256", + "source_profile_digest", "destination_profile_digest") + settings = ("source_model", "source_effort", "destination_model", "destination_effort") + if (not isinstance(value, Mapping) or set(value) != {*digests, *settings} or + any(not isinstance(value[k], str) or not BARE_SHA256_PATTERN.fullmatch(value[k]) for k in digests) or + any(not isinstance(value[k], str) or not value[k].strip() for k in settings)): + raise ValueError("checkpoint model transition facts are invalid") + + def codex_session_profile_digest( *, project: Path, @@ -419,8 +446,13 @@ def codex_session_profile_digest( reasoning_effort: str | None = None, sandbox: str | None = None, mcp_server: Mapping[str, Any] | None = None, + config_bytes: bytes | None = None, ) -> str: - """Bind a shared conversation to its workspace, provider home and settings.""" + """Bind a conversation to its workspace, provider home and settings. + + ``config_bytes`` compares inert captured settings at their qualified guest + paths. It does not install them or admit a changed active-session profile. + """ config = home / "config.toml" profile = { "workspace": str(project.resolve()), @@ -431,6 +463,7 @@ def codex_session_profile_digest( "sandbox": sandbox, "mcp_server": mcp_server, "config_digest": hashlib.sha256( + config_bytes if config_bytes is not None else config.read_bytes() if config.exists() else b"" ).hexdigest(), } diff --git a/loopx/hosts/codex_checkpoint.py b/loopx/hosts/codex_checkpoint.py index 37fba19ddc..5a6587aa8d 100644 --- a/loopx/hosts/codex_checkpoint.py +++ b/loopx/hosts/codex_checkpoint.py @@ -276,8 +276,42 @@ def _native_thread_ids(home: Path) -> set[str]: return set(ids) +def checkpoint_destination_config(config_path: Path, source_sha256: str, + destination_config: Path | None = None) -> tuple[bytes, dict[str, str] | None]: + """Verify sealed source settings and an explicit model/effort-only change. + + A destination is a new isolated receiver, never permission to change an + active session. Provider, tools, sandbox and all other configuration stay + semantically identical. Source bytes remain part of the sealed evidence. + """ + source = config_path.read_bytes() + if hashlib.sha256(source).hexdigest() != source_sha256: + raise ValueError("native checkpoint effective configuration differs") + if destination_config is None: + return source, None + if (not destination_config.is_absolute() or + destination_config.resolve(strict=True) != destination_config): + raise ValueError("checkpoint model configuration requires a physical absolute path") + target = destination_config.read_bytes() + original, changed = tomllib.loads(source.decode("utf-8")), tomllib.loads(target.decode("utf-8")) + keys = ("model", "model_reasoning_effort") + if any(not isinstance(config.get(key), str) or not config[key].strip() + for config in (original, changed) for key in keys): + raise ValueError("checkpoint model transition requires explicit model and reasoning effort") + if ({k: v for k, v in original.items() if k not in keys} != + {k: v for k, v in changed.items() if k not in keys}): + raise ValueError("checkpoint configuration transition may change only model and reasoning effort") + return target, { + "source_config_sha256": source_sha256, + "destination_config_sha256": hashlib.sha256(target).hexdigest(), + "source_model": original["model"], "source_effort": original["model_reasoning_effort"], + "destination_model": changed["model"], "destination_effort": changed["model_reasoning_effort"], + } + + def fork_native_checkpoint(env: dict[str, str], unit: Path, expected_sha256: str, - config_path: Path, *, timeout: float) -> dict[str, Any]: + config_path: Path, *, timeout: float, + destination_config: Path | None = None) -> dict[str, Any]: # The existing process lock serializes reconstruction in this isolated home. # The private attempt below describes one opaque native RPC, not authority # to admit a Goal, a work Turn, or a second fork after an uncertain response. @@ -293,11 +327,13 @@ def fork_native_checkpoint(env: dict[str, str], unit: Path, expected_sha256: str remaining = deadline - time.monotonic() if remaining <= 0: raise TimeoutError("native checkpoint reconstruction window closed") - return _fork_native_checkpoint(env, unit, expected_sha256, config_path, timeout=remaining) + return _fork_native_checkpoint(env, unit, expected_sha256, config_path, timeout=remaining, + **({"destination_config": destination_config} if destination_config is not None else {})) def _fork_native_checkpoint(env: dict[str, str], unit: Path, expected_sha256: str, - config_path: Path, *, timeout: float) -> dict[str, Any]: + config_path: Path, *, timeout: float, + destination_config: Path | None = None) -> dict[str, Any]: """Reconstruct one native branch in a new, fixed-path isolated environment. No model starts and no LoopX binding is registered. Existing session/Goal @@ -332,9 +368,8 @@ def _fork_native_checkpoint(env: dict[str, str], unit: Path, expected_sha256: st recovering = home.exists() if recovering and (not attempt_path.is_file() or attempt_path.resolve() != attempt_path): raise ValueError("native checkpoint destination home must be new or retain its original fork attempt") - config = config_path.read_bytes() - if hashlib.sha256(config).hexdigest() != native["config_sha256"]: - raise ValueError("native checkpoint effective configuration differs") + config, transition = checkpoint_destination_config( + config_path, native["config_sha256"], destination_config) version = subprocess.check_output([env["CODEX_BIN"], "--version"], env=env, text=True, encoding="utf-8", timeout=max(.1, deadline - time.monotonic())).strip() if version != native["host_version"]: @@ -368,6 +403,8 @@ def _fork_native_checkpoint(env: dict[str, str], unit: Path, expected_sha256: st **cursor_basis, "config_sha256": native["config_sha256"], "destination_goal_id": env.get("LOOPX_GOAL_ID"), "destination_agent_id": env.get("LOOPX_AGENT_ID"), "guest_workspace": env["LOOPX_PROJECT"], "guest_home": str(home)} + if transition is not None: + basis["model_transition"] = transition attempt = None if recovering: attempt = json.loads(attempt_path.read_bytes()) @@ -443,11 +480,13 @@ def request(method, params): raise ValueError("native checkpoint reconstructed history differs") return {"session_id": session, **cursor_basis, "visible_history_sha256": native["visible_history_sha256"], "history_verified": True, + **({"model_transition": transition} if transition is not None else {}), "model_calls": 0, "execution_authority_granted": False, "continuation_qualified": False} def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sha256: str, - config_path: Path, *, sandbox: str, timeout: float) -> dict[str, Any]: + config_path: Path, *, sandbox: str, timeout: float, + destination_config: Path | None = None) -> dict[str, Any]: """Explicit controller admission of a verified fork into a destination Goal. Existing native fork, Goal lifetime and session owners retain their rules. @@ -461,7 +500,9 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh from loopx.control_plane.projects.registry_codec import ( project_registry_transaction, ) - from loopx.control_plane.turn_driver.codex_sessions import select_codex_cli_session + from loopx.control_plane.turn_driver.codex_sessions import ( + codex_session_profile_digest, select_codex_cli_session, + ) from loopx.control_plane.turn_driver.codex_agent_session import CodexAgentSession if not math.isfinite(timeout) or timeout <= 0: @@ -478,8 +519,29 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh raise ValueError("checkpoint session admission requires its captured source session profile") if tomllib.loads(config_path.read_text(encoding="utf-8")).get("features", {}).get("goals") is not False: raise ValueError("checkpoint session admission requires native automatic Goals explicitly disabled") - if hashlib.sha256(config_path.read_bytes()).hexdigest() != metadata["native_session"]["config_sha256"]: - raise ValueError("native checkpoint effective configuration differs") + config, transition = checkpoint_destination_config( + config_path, metadata["native_session"]["config_sha256"], destination_config) + target_profile = source_profile + if transition is not None: + if (env["MODEL_NAME"] != transition["destination_model"] or + env["REASONING_EFFORT"] != transition["destination_effort"]): + raise ValueError("checkpoint destination model differs from the current launch") + binary = shutil.which(env["CODEX_BIN"]) + if binary is None: + raise ValueError("Codex CLI executable is unavailable") + observed_source = codex_session_profile_digest( + project=Path(env["LOOPX_PROJECT"]), codex_bin=binary, + home=Path(env["CODEX_HOME"]), sandbox=sandbox, + model=transition["source_model"], reasoning_effort=transition["source_effort"], + config_bytes=config_path.read_bytes()) + if observed_source != source_profile: + raise ValueError("checkpoint source session profile changed") + target_profile = codex_session_profile_digest( + project=Path(env["LOOPX_PROJECT"]), codex_bin=binary, + home=Path(env["CODEX_HOME"]), sandbox=sandbox, + model=env["MODEL_NAME"], reasoning_effort=env["REASONING_EFFORT"], config_bytes=config) + transition = {**transition, "source_profile_digest": source_profile, + "destination_profile_digest": target_profile} source = metadata.get("loopx_state", {}) lineage = {"goal_id": env["LOOPX_GOAL_ID"], "agent_id": env["LOOPX_AGENT_ID"]} if (not source.get("goal_id") or not source.get("agent_id") or @@ -510,8 +572,10 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh raise ValueError("destination session binding already exists without this checkpoint admission") if binding.get("checkpoint_artifact_root") not in (None, str(unit)): raise ValueError("destination checkpoint source root differs from the original admission") - if wake.digest != source_profile: + if wake.digest != target_profile: raise ValueError("checkpoint source session profile changed") + if transition is not None and binding.get("checkpoint_model_transition") != transition: + raise ValueError("checkpoint model transition differs from its original admission") native = metadata["native_session"] observed = native_session_checkpoint(env, binding["session_id"], timeout=max(.1, deadline - time.monotonic()), @@ -519,15 +583,18 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh cursor_key = "interrupted_turn_id" if native.get("interrupted_guard") is not None else "completed_turn_id" keys = (cursor_key, "turn_ids", "visible_history_sha256", "host_version", "guest_home", "guest_workspace", "config_sha256") - if binding["session_id"] == native["session_id"] or any(observed[key] != native[key] for key in keys): + expected = {**native, "config_sha256": hashlib.sha256(config).hexdigest()} + if binding["session_id"] == native["session_id"] or any(observed[key] != expected[key] for key in keys): raise ValueError("checkpoint admitted native session advanced or differs; use its existing continuation") return {"session_id": binding["session_id"], cursor_key: observed[cursor_key], "visible_history_sha256": observed["visible_history_sha256"], "history_verified": True, "model_calls": 0, "execution_authority_granted": False, "continuation_qualified": False, "destination_goal_ref": wake.goal_ref, "destination_agent_id": lineage["agent_id"], - "session_binding_admitted": True, "session_profile_digest": wake.digest} + "session_binding_admitted": True, "session_profile_digest": wake.digest, + **({"model_transition": transition} if transition is not None else {})} result = fork_native_checkpoint(env, unit, expected_sha256, config_path, - timeout=max(.1, deadline - time.monotonic())) + timeout=max(.1, deadline - time.monotonic()), + **({"destination_config": destination_config} if destination_config is not None else {})) receipt: dict[str, Any] = {} wake = CodexAgentSession(env, context="resume", sandbox=sandbox, receipt=receipt) # Hold the existing registry owner's fence through the session write. This @@ -537,8 +604,13 @@ def admit_native_checkpoint_session(env: dict[str, str], unit: Path, expected_sh require_registered_agent_id(registry_path=registry, goal_id=lineage["goal_id"], agent_id=lineage["agent_id"]) if current != goal: raise ValueError("destination Goal configuration changed during native checkpoint reconstruction") - wake._bind_verified_fork(result["session_id"], source_profile, expected_sha256, str(unit)) - return {**result, "destination_goal_ref": wake.goal_ref, "destination_agent_id": lineage["agent_id"], + # The source profile was independently checked against the sealed + # settings above. Bind the explicitly verified destination profile; + # ordinary session resumes still require an exact profile match. + wake._bind_verified_fork(result["session_id"], target_profile, expected_sha256, str(unit), + **({"model_transition": transition} if transition is not None else {})) + return {**result, **({"model_transition": transition} if transition is not None else {}), + "destination_goal_ref": wake.goal_ref, "destination_agent_id": lineage["agent_id"], "session_binding_admitted": True, "session_profile_digest": wake.digest} @@ -551,6 +623,8 @@ def main(argv=None) -> None: parser.add_argument("--root", type=Path, required=True) parser.add_argument("--expected-sha256", required=True) parser.add_argument("--config-path", type=Path, required=True) + parser.add_argument("--destination-model-config", type=Path, + help="Explicit new-receiver config changing only model and reasoning effort") parser.add_argument("--sandbox", choices=CODEX_CLI_SANDBOXES, required=True) parser.add_argument("--timeout-seconds", type=float, default=30) args = parser.parse_args(argv) @@ -560,7 +634,9 @@ def main(argv=None) -> None: if missing: parser.error("Explicit destination controller environment is missing: " + ", ".join(missing)) result = admit_native_checkpoint_session(dict(os.environ), args.root, args.expected_sha256, - args.config_path, sandbox=args.sandbox, timeout=args.timeout_seconds) + args.config_path, sandbox=args.sandbox, timeout=args.timeout_seconds, + **({"destination_config": args.destination_model_config} + if args.destination_model_config is not None else {})) print(json.dumps(result)) diff --git a/loopx/hosts/research_checkpoint.py b/loopx/hosts/research_checkpoint.py index 62e3332eea..f45d6f84a5 100644 --- a/loopx/hosts/research_checkpoint.py +++ b/loopx/hosts/research_checkpoint.py @@ -135,21 +135,34 @@ def _handoff(env, payload): -def _context_session(env, metadata, timeout=30): +def _context_session(env, metadata, timeout=30, *, checkpoint_sha256=None): """Read the admitted native branch before publishing captured Todo context.""" from loopx.control_plane.turn_driver.codex_agent_session import CodexAgentSession from loopx.hosts.codex_checkpoint import native_session_checkpoint wake = CodexAgentSession(env, context="resume", sandbox=env["LOOPX_CODEX_SANDBOX"], receipt={}) native = metadata["native_session"] + expected_profile = metadata.get("session", {}).get("session_profile_digest") + expected_config = native["config_sha256"] + binding = wake.binding or {} + transition = binding.get("checkpoint_model_transition") + if transition is not None: + from loopx.control_plane.turn_driver.codex_sessions import validate_checkpoint_model_transition + validate_checkpoint_model_transition(transition) + if (not checkpoint_sha256 or binding.get("checkpoint_artifact_sha256") != checkpoint_sha256 or + transition["source_profile_digest"] != expected_profile or + transition["source_config_sha256"] != expected_config): + raise ValueError("context adoption model transition does not match the captured source") + expected_profile = transition["destination_profile_digest"] + expected_config = transition["destination_config_sha256"] if (not wake.session_id or wake.session_id == native["session_id"] or - wake.digest != metadata.get("session", {}).get("session_profile_digest")): + wake.digest != expected_profile): raise ValueError("context adoption requires the admitted destination session profile") observed = native_session_checkpoint(env, wake.session_id, timeout=timeout, interrupted_guard=native.get("interrupted_guard"))["metadata"] cursor_key = "interrupted_turn_id" if native.get("interrupted_guard") is not None else "completed_turn_id" for key in ("visible_history_sha256", "turn_ids", cursor_key, "host_version", "config_sha256"): - if observed[key] != native[key]: + if observed[key] != (expected_config if key == "config_sha256" else native[key]): raise ValueError("destination native history or configuration moved beyond the captured context") return wake.session_id @@ -189,7 +202,7 @@ def verify_artifact(): read_timeout() # A late successful read cannot authorize the next write. def read_session(): - session = _context_session(env, metadata, read_timeout()) + session = _context_session(env, metadata, read_timeout(), checkpoint_sha256=digest) read_timeout() return session