mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
fix(codex): preserve wrapped sessions and recover state (#2160)
## Description Closes #2159. Codex wrappers currently launch against a disposable `CODEX_HOME`, so session state created during a wrapped run can disappear when that temporary directory is removed. This change launches Codex against its durable home, keeps proxy routing process-local, and adds recovery for retained temporary homes and pinned recovery sources. ## Type of Change - [x] Bug fix (non-breaking change that fixes an issue) - [ ] New feature (non-breaking change that adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [x] Documentation update - [ ] Performance improvement - [ ] Code refactoring (no functional changes) ## Changes Made - Launch Codex against its durable `CODEX_HOME` and apply routing through process-local config overrides after the actual proxy port is resolved. - Preserve custom provider identity and reject providers that cannot be redirected safely. - Detect dangling temporary Codex homes before interactive wraps and offer recovery. - Add `headroom recover codex` with automatic discovery, repeatable `--source`, preview, confirmation, retained backups, and rollback on failure. - Search Python's temp root, `$TMPDIR`, `/tmp`, `/private/tmp`, and macOS `/private/var/folders/*/*/T` for retained `headroom-codex-home-*` directories. - Reuse `source-pinned/` copies left by interrupted or failed recovery attempts after the original temporary home has disappeared. - Report deleted temporary homes still referenced by SQLite rollout paths without treating paths pasted into prompts or errors as filesystem evidence. - Audit the durable thread index, rollout files, and history when no source remains, including indexed chat counts and history-only orphan records. - Normalize legacy localhost `headroom` providers in both SQLite thread rows and rollout `session_meta`, including retries after an earlier broken recovery, while preserving user-defined remote providers named `headroom`. - Merge compatible config, JSONL, rollout, SQLite, credential, and regular-file state without propagating deletions or runtime artifacts. - Rewrite recovered thread rollout paths to the durable home and restore legacy Headroom thread providers to the active provider. - Validate SQLite schemas, SQLx migration checksums, integrity, and foreign keys, and quarantine malformed JSONL. - Preserve failed targets with an atomic rename before rollback, avoiding recursive-deletion races with live SQLite runtime files. - Document discovery, migration, retained backups, rollback behavior, and the limits of deleted-source recovery. ## Testing - [x] Unit tests pass (`pytest`) - [x] Linting passes (`ruff check`) - [x] Type checking passes (`mypy`) - [x] New tests added for new functionality - [x] Manual testing performed in isolated Docker containers ### Test Output ```text $ uv run pytest tests/test_cli/test_wrap_codex.py tests/test_cli/test_recover_codex.py -q 122 passed $ uv run ruff check headroom/cli/recover.py headroom/providers/codex/recovery.py tests/test_cli/test_recover_codex.py All checks passed! $ uv run ruff format --check headroom/cli/recover.py headroom/providers/codex/recovery.py tests/test_cli/test_recover_codex.py 3 files already formatted $ uv run mypy headroom/cli/recover.py headroom/providers/codex/recovery.py Success: no issues found in 2 source files ``` All validation ran in `ghcr.io/astral-sh/uv:python3.12-bookworm` against a writable disposable copy of a read-only source mount. Codex was not installed or launched, and no real user Codex state was read or modified. The tests cover multi-root discovery, deleted-reference reporting, retained pinned-source recovery, durable SQLite path relocation, SQLite and rollout provider normalization, idempotent repair after an earlier broken recovery, remote provider preservation, unrelated dangling target rows, backup retention, atomic rollback, malformed-state quarantine, SQLite validation, and Windows-safe handle closure. The repository shim E2E was not launched locally because this recovery work intentionally avoids launching Codex. Upstream CI exercises wrapper E2E in isolated environments. ## Real Behavior Proof - Environment: `ghcr.io/astral-sh/uv:python3.12-bookworm`, Python 3.12, a writable disposable checkout copied from a read-only source mount, at head `2d89ecec`. - Exact command / steps: Run `pytest -q tests/test_cli/test_wrap_codex.py tests/test_cli/test_recover_codex.py`, then run `ruff check` and `ruff format --check` against `headroom/cli/wrap.py`, `headroom/cli/recover.py`, `headroom/providers/codex/recovery.py`, `tests/test_cli/test_wrap_codex.py`, and `tests/test_cli/test_recover_codex.py`. - Observed result: `122 passed in 10.08s`; Ruff reported `All checks passed!` and `5 files already formatted`. - Not tested: Launching a real Codex process or modifying a real user `CODEX_HOME`; these were intentionally excluded to protect live user state. ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review ## Checklist - [x] My code follows the project's style guidelines - [x] I have performed a self-review of my code - [x] I have commented my code where the behavior is hard to understand - [x] I have made corresponding documentation changes - [x] My changes generate no new warnings - [x] I have added tests that prove the fix is effective - [x] New and existing focused unit tests pass with my changes - [x] I have updated `CHANGELOG.md` if applicable ## Additional Notes The temporary-home behavior was introduced by #1507 in `ad9d086f43`. Related context: #730, #731, #961, #1034, #1050, #1349, #1853, #1889, #2103, and #2104. A temporary home that macOS or `TemporaryDirectory` already deleted cannot be reconstructed unless a retained `source-pinned/` copy exists. Recovery identifies genuine dangling SQLite paths, audits surviving durable history, and recovers any retained pinned source it can find. Prompt text without a rollout cannot reconstruct a full transcript. The unchecked changelog item is not applicable because this repository does not require a changelog entry for this fix. --------- Co-authored-by: JerrettDavis <mxjerrett@gmail.com>
This commit is contained in:
Rudimar Ronsoni
GitHub
parent
57e8dcb425
commit
dec60de976
13 changed files with 2195 additions and 250 deletions
|
|
@ -10,16 +10,19 @@ way a user would from the shell.
|
|||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
import socket
|
||||
import sqlite3
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
import tomllib
|
||||
from click.testing import CliRunner
|
||||
|
||||
if sys.version_info >= (3, 11):
|
||||
import tomllib
|
||||
else: # pragma: no cover - exercised in the Python 3.10 test job
|
||||
import tomli as tomllib
|
||||
|
||||
from headroom.cli import wrap as wrap_mod
|
||||
from headroom.cli.main import main
|
||||
from headroom.mcp_registry.install import build_headroom_spec
|
||||
|
|
@ -840,8 +843,6 @@ class TestInjectAvoidsDuplicateTopLevelKeys:
|
|||
def test_inject_does_not_create_duplicate_model_provider(
|
||||
self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
import tomllib # Python 3.11+ stdlib
|
||||
|
||||
_set_test_home(monkeypatch, tmp_path)
|
||||
config_dir = tmp_path / ".codex"
|
||||
config_dir.mkdir()
|
||||
|
|
@ -903,8 +904,6 @@ class TestInjectAvoidsDuplicateTopLevelKeys:
|
|||
self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
"""Idempotent re-wrap on a config that already has top-level keys."""
|
||||
import tomllib
|
||||
|
||||
_set_test_home(monkeypatch, tmp_path)
|
||||
config_dir = tmp_path / ".codex"
|
||||
config_dir.mkdir()
|
||||
|
|
@ -939,8 +938,6 @@ class TestInjectAvoidsDuplicateTopLevelKeys:
|
|||
self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
"""Existing headroom provider table must not create duplicate TOML keys."""
|
||||
import tomllib # Python 3.11+ stdlib
|
||||
|
||||
_set_test_home(monkeypatch, tmp_path)
|
||||
config_dir = tmp_path / ".codex"
|
||||
config_dir.mkdir()
|
||||
|
|
@ -1055,67 +1052,7 @@ def test_wrap_codex_prepare_only_respects_codex_home(
|
|||
assert not (tmp_path / ".codex" / "config.toml").exists()
|
||||
|
||||
|
||||
def test_codex_session_home_overlay_seeds_active_home_and_cleans_up(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
_set_test_home(monkeypatch, tmp_path)
|
||||
codex_home = tmp_path / "custom-codex-home"
|
||||
codex_home.mkdir()
|
||||
monkeypatch.setenv("CODEX_HOME", str(codex_home))
|
||||
|
||||
config_file = codex_home / "config.toml"
|
||||
auth_file = codex_home / "auth.json"
|
||||
original_config = '[profiles.default]\nmodel = "gpt-4o"\n'
|
||||
original_auth = '{"auth_mode": "apikey"}'
|
||||
config_file.write_text(original_config, encoding="utf-8")
|
||||
auth_file.write_text(original_auth, encoding="utf-8")
|
||||
|
||||
with wrap_mod._codex_session_home_overlay() as session_home:
|
||||
seeded_config = (session_home / "config.toml").read_text(encoding="utf-8")
|
||||
seeded_auth = (session_home / "auth.json").read_text(encoding="utf-8")
|
||||
assert seeded_config == original_config
|
||||
assert seeded_auth == original_auth
|
||||
(session_home / "config.toml").write_text('model_provider = "headroom"\n', encoding="utf-8")
|
||||
assert config_file.read_text(encoding="utf-8") == original_config
|
||||
|
||||
assert not session_home.exists()
|
||||
assert config_file.read_text(encoding="utf-8") == original_config
|
||||
assert auth_file.read_text(encoding="utf-8") == original_auth
|
||||
|
||||
|
||||
@pytest.mark.skipif(
|
||||
sys.platform == "win32" or not hasattr(socket, "AF_UNIX"),
|
||||
reason="requires POSIX Unix domain sockets",
|
||||
)
|
||||
def test_codex_session_home_overlay_skips_unix_sockets(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
_set_test_home(monkeypatch, tmp_path)
|
||||
codex_home = tmp_path / "custom-codex-home"
|
||||
socket_dir = codex_home / "vendor_imports" / "skills" / ".git"
|
||||
socket_dir.mkdir(parents=True)
|
||||
monkeypatch.setenv("CODEX_HOME", str(codex_home))
|
||||
monkeypatch.chdir(codex_home)
|
||||
|
||||
head_file = socket_dir / "HEAD"
|
||||
head_file.write_text("ref: refs/heads/main\n", encoding="utf-8")
|
||||
socket_file = socket_dir / "fsmonitor--daemon.ipc"
|
||||
relative_socket_file = socket_file.relative_to(codex_home)
|
||||
|
||||
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as fsmonitor_socket:
|
||||
fsmonitor_socket.bind(str(relative_socket_file))
|
||||
|
||||
with wrap_mod._codex_session_home_overlay() as session_home:
|
||||
session_socket_dir = session_home / socket_dir.relative_to(codex_home)
|
||||
assert (session_socket_dir / "HEAD").read_text(encoding="utf-8") == (
|
||||
"ref: refs/heads/main\n"
|
||||
)
|
||||
assert not (session_socket_dir / socket_file.name).exists()
|
||||
|
||||
assert socket_file.is_socket()
|
||||
|
||||
|
||||
def test_wrap_codex_launch_uses_session_scoped_codex_home(
|
||||
def test_wrap_codex_launch_uses_durable_codex_home(
|
||||
runner: CliRunner, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
_set_test_home(monkeypatch, tmp_path)
|
||||
|
|
@ -1131,7 +1068,7 @@ def test_wrap_codex_launch_uses_session_scoped_codex_home(
|
|||
auth_file.write_text(original_auth, encoding="utf-8")
|
||||
|
||||
launch_env: dict[str, str] = {}
|
||||
session_home_seen: list[Path] = []
|
||||
rollout = codex_home / "sessions" / "2026" / "07" / "14" / "rollout-thread.jsonl"
|
||||
|
||||
def fake_launch(
|
||||
*,
|
||||
|
|
@ -1147,15 +1084,9 @@ def test_wrap_codex_launch_uses_session_scoped_codex_home(
|
|||
del args, port, no_proxy, tool_label, env_vars_display, kwargs
|
||||
assert binary == "/fake/codex"
|
||||
launch_env.update(env)
|
||||
session_home = Path(env["CODEX_HOME"])
|
||||
session_home_seen.append(session_home)
|
||||
assert session_home.exists()
|
||||
seeded_config = (session_home / "config.toml").read_text(encoding="utf-8")
|
||||
assert original_config in seeded_config
|
||||
assert 'model_provider = "headroom"' in seeded_config
|
||||
assert 'base_url = "http://127.0.0.1:8787/v1"' in seeded_config
|
||||
assert "[mcp_servers.headroom]" in seeded_config
|
||||
assert (session_home / "auth.json").read_text(encoding="utf-8") == original_auth
|
||||
assert Path(env["CODEX_HOME"]) == codex_home
|
||||
rollout.parent.mkdir(parents=True)
|
||||
rollout.write_text('{"type":"session_meta"}\n', encoding="utf-8")
|
||||
|
||||
with patch("headroom.cli.wrap._ensure_rtk_binary", return_value=None):
|
||||
with patch(
|
||||
|
|
@ -1176,48 +1107,94 @@ def test_wrap_codex_launch_uses_session_scoped_codex_home(
|
|||
)
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
assert session_home_seen
|
||||
assert launch_env["CODEX_HOME"] == str(session_home_seen[0])
|
||||
assert launch_env["CODEX_HOME"] == str(codex_home)
|
||||
assert launch_env["OPENAI_BASE_URL"] == "http://127.0.0.1:8787/v1"
|
||||
assert config_file.read_text(encoding="utf-8") == original_config
|
||||
persisted_config = config_file.read_text(encoding="utf-8")
|
||||
assert original_config in persisted_config
|
||||
assert "[mcp_servers.headroom]" in persisted_config
|
||||
assert 'model_provider = "headroom"' not in persisted_config
|
||||
assert "[model_providers.headroom]" not in persisted_config
|
||||
assert auth_file.read_text(encoding="utf-8") == original_auth
|
||||
assert not session_home_seen[0].exists()
|
||||
assert rollout.read_text(encoding="utf-8") == '{"type":"session_meta"}\n'
|
||||
|
||||
|
||||
def test_wrap_codex_launches_use_distinct_session_homes_per_port(
|
||||
def test_codex_session_launch_settings_keep_routing_process_local(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
_set_test_home(monkeypatch, tmp_path)
|
||||
codex_home = tmp_path / "custom-codex-home"
|
||||
codex_home.mkdir()
|
||||
monkeypatch.setenv("CODEX_HOME", str(codex_home))
|
||||
monkeypatch.setattr(wrap_mod, "_project_name_from_cwd", lambda: None)
|
||||
config_file = codex_home / "config.toml"
|
||||
original_config = 'model = "gpt-5"\n'
|
||||
config_file.write_text(original_config, encoding="utf-8")
|
||||
|
||||
args, env, display = wrap_mod._codex_session_launch_settings(
|
||||
port=9898,
|
||||
codex_args=("exec", "hello"),
|
||||
environ={"CODEX_HOME": str(codex_home)},
|
||||
)
|
||||
|
||||
assert args == (
|
||||
"--config",
|
||||
'openai_base_url="http://127.0.0.1:9898/v1"',
|
||||
"exec",
|
||||
"hello",
|
||||
)
|
||||
assert env["CODEX_HOME"] == str(codex_home)
|
||||
assert env["OPENAI_BASE_URL"] == "http://127.0.0.1:9898/v1"
|
||||
assert display == ["OPENAI_BASE_URL=http://127.0.0.1:9898/v1"]
|
||||
assert config_file.read_text(encoding="utf-8") == original_config
|
||||
|
||||
|
||||
def test_codex_session_launch_settings_preserve_custom_provider_identity(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
_set_test_home(monkeypatch, tmp_path)
|
||||
codex_home = tmp_path / "custom-codex-home"
|
||||
codex_home.mkdir()
|
||||
monkeypatch.setenv("CODEX_HOME", str(codex_home))
|
||||
monkeypatch.setattr(wrap_mod, "_project_name_from_cwd", lambda: None)
|
||||
config_file = codex_home / "config.toml"
|
||||
original_config = (
|
||||
'[profiles.work]\nmodel_provider = "company"\n\n'
|
||||
'[model_providers.company]\nbase_url = "https://api.example.test/v1"\n'
|
||||
)
|
||||
config_file.write_text(original_config, encoding="utf-8")
|
||||
|
||||
args, env, _ = wrap_mod._codex_session_launch_settings(
|
||||
port=9898,
|
||||
codex_args=("--profile", "work"),
|
||||
environ={"CODEX_HOME": str(codex_home)},
|
||||
)
|
||||
|
||||
assert "model_provider=headroom" not in " ".join(args)
|
||||
assert '"model_providers"."company"."base_url"="http://127.0.0.1:9898/v1"' in args
|
||||
assert env[wrap_mod._UPSTREAM_BASE_URL_ENV_VAR] == "https://api.example.test/v1"
|
||||
assert config_file.read_text(encoding="utf-8") == original_config
|
||||
|
||||
|
||||
def test_wrap_codex_rejects_custom_provider_without_upstream_base_url(
|
||||
runner: CliRunner, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
_set_test_home(monkeypatch, tmp_path)
|
||||
codex_home = tmp_path / "custom-codex-home"
|
||||
codex_home.mkdir()
|
||||
monkeypatch.setenv("CODEX_HOME", str(codex_home))
|
||||
(codex_home / "config.toml").write_text(
|
||||
'model_provider = "company"\n[model_providers.company]\nname = "Company"\n',
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
config_file = codex_home / "config.toml"
|
||||
auth_file = codex_home / "auth.json"
|
||||
original_config = '[profiles.default]\nmodel = "gpt-4o"\n'
|
||||
original_auth = '{"auth_mode": "apikey"}'
|
||||
config_file.write_text(original_config, encoding="utf-8")
|
||||
auth_file.write_text(original_auth, encoding="utf-8")
|
||||
|
||||
launch_records: list[tuple[int, Path, str]] = []
|
||||
|
||||
def fake_launch(
|
||||
*,
|
||||
binary: str,
|
||||
args: tuple,
|
||||
env: dict[str, str],
|
||||
port: int,
|
||||
no_proxy: bool,
|
||||
tool_label: str,
|
||||
env_vars_display: list[str],
|
||||
**kwargs: object,
|
||||
) -> None:
|
||||
del args, no_proxy, tool_label, env_vars_display, kwargs
|
||||
assert binary == "/fake/codex"
|
||||
session_home = Path(env["CODEX_HOME"])
|
||||
assert session_home.exists()
|
||||
launch_records.append(
|
||||
(port, session_home, (session_home / "config.toml").read_text(encoding="utf-8"))
|
||||
def fake_launch(**kwargs: object) -> None:
|
||||
configure_launch = kwargs["configure_launch"]
|
||||
assert callable(configure_launch)
|
||||
configure_launch(
|
||||
8787,
|
||||
kwargs["args"],
|
||||
kwargs["env"],
|
||||
kwargs["env_vars_display"],
|
||||
)
|
||||
|
||||
with patch("headroom.cli.wrap._ensure_rtk_binary", return_value=None):
|
||||
|
|
@ -1226,25 +1203,72 @@ def test_wrap_codex_launches_use_distinct_session_homes_per_port(
|
|||
side_effect=lambda cmd: "/fake/codex" if cmd == "codex" else None,
|
||||
):
|
||||
with patch("headroom.cli.wrap._launch_tool", side_effect=fake_launch):
|
||||
first = runner.invoke(
|
||||
result = runner.invoke(
|
||||
main,
|
||||
["wrap", "codex", "--port", "8787", "--no-tokensave", "--no-serena"],
|
||||
)
|
||||
second = runner.invoke(
|
||||
main,
|
||||
["wrap", "codex", "--port", "9898", "--no-tokensave", "--no-serena"],
|
||||
[
|
||||
"wrap",
|
||||
"codex",
|
||||
"--port",
|
||||
"8787",
|
||||
"--no-rtk",
|
||||
"--no-mcp",
|
||||
"--no-tokensave",
|
||||
"--no-serena",
|
||||
],
|
||||
)
|
||||
|
||||
assert first.exit_code == 0, first.output
|
||||
assert second.exit_code == 0, second.output
|
||||
assert len(launch_records) == 2
|
||||
assert launch_records[0][1] != launch_records[1][1]
|
||||
assert 'base_url = "http://127.0.0.1:8787/v1"' in launch_records[0][2]
|
||||
assert 'base_url = "http://127.0.0.1:9898/v1"' in launch_records[1][2]
|
||||
assert config_file.read_text(encoding="utf-8") == original_config
|
||||
assert auth_file.read_text(encoding="utf-8") == original_auth
|
||||
assert not launch_records[0][1].exists()
|
||||
assert not launch_records[1][1].exists()
|
||||
assert result.exit_code != 0
|
||||
assert "custom provider 'company' has no upstream base_url" in result.output
|
||||
|
||||
|
||||
def test_wrap_codex_routes_model_provider_selected_by_config_argument(
|
||||
runner: CliRunner, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
_set_test_home(monkeypatch, tmp_path)
|
||||
codex_home = tmp_path / "custom-codex-home"
|
||||
codex_home.mkdir()
|
||||
monkeypatch.setenv("CODEX_HOME", str(codex_home))
|
||||
monkeypatch.setattr(wrap_mod, "_project_name_from_cwd", lambda: None)
|
||||
(codex_home / "config.toml").write_text(
|
||||
'[model_providers.company]\nbase_url = "https://api.example.test/v1"\n',
|
||||
encoding="utf-8",
|
||||
)
|
||||
configured_env: dict[str, str] = {}
|
||||
|
||||
def fake_launch(**kwargs: object) -> None:
|
||||
configure_launch = kwargs["configure_launch"]
|
||||
assert callable(configure_launch)
|
||||
_, env, _ = configure_launch(
|
||||
8787,
|
||||
kwargs["args"],
|
||||
kwargs["env"],
|
||||
kwargs["env_vars_display"],
|
||||
)
|
||||
configured_env.update(env)
|
||||
|
||||
with patch("headroom.cli.wrap._ensure_rtk_binary", return_value=None):
|
||||
with patch(
|
||||
"headroom.cli.wrap.shutil.which",
|
||||
side_effect=lambda cmd: "/fake/codex" if cmd == "codex" else None,
|
||||
):
|
||||
with patch("headroom.cli.wrap._launch_tool", side_effect=fake_launch):
|
||||
result = runner.invoke(
|
||||
main,
|
||||
[
|
||||
"wrap",
|
||||
"codex",
|
||||
"--no-rtk",
|
||||
"--no-mcp",
|
||||
"--no-tokensave",
|
||||
"--no-serena",
|
||||
"--",
|
||||
"--config",
|
||||
'model_provider="company"',
|
||||
],
|
||||
)
|
||||
|
||||
assert result.exit_code == 0, result.output
|
||||
assert configured_env[wrap_mod._UPSTREAM_BASE_URL_ENV_VAR] == ("https://api.example.test/v1")
|
||||
|
||||
|
||||
def test_wrap_codex_injects_rtk_globally_without_changing_project_agents(
|
||||
|
|
@ -1968,22 +1992,28 @@ class TestCodexLaunchExportsCustomUpstream:
|
|||
the wrong host (regression of #1614)."""
|
||||
|
||||
def _launch_env(self, monkeypatch, tmp_path, *, custom_upstream):
|
||||
from contextlib import contextmanager
|
||||
|
||||
captured: dict = {}
|
||||
|
||||
monkeypatch.setattr(wrap_mod.shutil, "which", lambda name: "/usr/bin/codex")
|
||||
monkeypatch.setattr(wrap_mod, "_codex_home_dir", lambda: tmp_path)
|
||||
monkeypatch.setattr(wrap_mod, "_offer_dangling_codex_recovery", lambda active_home: None)
|
||||
monkeypatch.setattr(wrap_mod, "_prepare_codex_wrap_state", lambda **kwargs: None)
|
||||
if custom_upstream:
|
||||
(tmp_path / "config.toml").write_text(
|
||||
"\n".join(
|
||||
(
|
||||
'model_provider = "gateway"',
|
||||
"[model_providers.gateway]",
|
||||
f'base_url = "{custom_upstream}"',
|
||||
)
|
||||
)
|
||||
+ "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
@contextmanager
|
||||
def _fake_overlay():
|
||||
yield tmp_path / "session"
|
||||
|
||||
monkeypatch.setattr(wrap_mod, "_codex_session_home_overlay", _fake_overlay)
|
||||
# Stand in for the heavy prepare step; only its return value matters here.
|
||||
monkeypatch.setattr(wrap_mod, "_prepare_codex_wrap_state", lambda **kwargs: custom_upstream)
|
||||
|
||||
def _fake_launch(*, env, **kwargs):
|
||||
def _fake_launch(*, env, port, configure_launch, args=(), env_vars_display=(), **kwargs):
|
||||
if configure_launch is not None:
|
||||
_args, env, _display = configure_launch(port, args, env, list(env_vars_display))
|
||||
captured["env"] = env
|
||||
|
||||
monkeypatch.setattr(wrap_mod, "_launch_tool", _fake_launch)
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue