from __future__ import annotations import base64 import json import os import re import subprocess import time from pathlib import Path import pytest import command_router import remote_dispatch from health_client import AccountSnapshot, HealthStatus def write_rules(base_dir: Path, payload: dict[str, object]) -> None: tray_dir = base_dir / ".local/state/overdeck/systray/runtime" tray_dir.mkdir(parents=True, exist_ok=True) (tray_dir / payload["filename"]).write_text( json.dumps(payload["rules"]), encoding="utf-8", ) def write_health(base_dir: Path, payload: dict[str, dict[str, object]], filename: str) -> None: tray_dir = base_dir / ".local/state/overdeck/systray/runtime" tray_dir.mkdir(parents=True, exist_ok=True) (tray_dir / filename).write_text(json.dumps(payload), encoding="utf-8") @pytest.fixture def home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: monkeypatch.setattr(command_router.Path, "home", lambda: tmp_path) return tmp_path def test_result_transport_failure_overrides_only_successful_agent_exit() -> None: assert command_router._result_transport_exit_code(0, "result missing") == remote_dispatch.EXIT_MIRROR assert command_router._result_transport_exit_code(17, "result missing") == 17 assert command_router._result_transport_exit_code(0, None) == 0 def test_codex_router_rejects_missing_project_account_using_known_slugs( home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: write_rules( home, { "filename": "routing_rules.json", "rules": { "projects": {"zync.is": "avi"}, "default": "rafa", "fallback_chain": ["roy"], "fallback_trigger": "broken_or_quota_exhausted", "quota_exhausted_threshold_pct": 100, }, }, ) for slug in ("rafa", "roy"): (home / ".local/state/overdeck/systray/runtime" / "accounts" / slug / "CODEX_HOME").mkdir(parents=True) write_health( home, { "rafa": {"status": "ok", "primary_used_pct": 10, "secondary_used_pct": 20}, }, "health_cache.json", ) monkeypatch.setattr(command_router, "detect_project_name", lambda cwd=None: "zync.is") router = command_router.CommandRouter( command_router.CodexAdapter(), detect_project_name=command_router.detect_project_name, ) exec_called = False def fake_execvpe(file: str, args: list[str], env: dict[str, str]) -> None: nonlocal exec_called exec_called = True monkeypatch.setattr(command_router.os, "execvpe", fake_execvpe) exit_code = router.main(["cdx", "chat"]) captured = capsys.readouterr() assert exit_code == 1 assert exec_called is False assert captured.err.strip() == "cdx: no healthy account available: avi=unknown-account" def test_cld_router_prefixes_errors_without_cdx_leak( home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: write_rules( home, { "filename": "claude_routing_rules.json", "rules": { "projects": {}, "default": "avi", "fallback_chain": ["roy"], "fallback_trigger": "broken_or_quota_exhausted", "quota_exhausted_threshold_pct": 100, }, }, ) monkeypatch.setattr(command_router, "detect_project_name", lambda cwd=None: "unknown") write_health(home, {}, "claude_health_cache.json") router = command_router.CommandRouter( command_router.ClaudeAdapter(), detect_project_name=command_router.detect_project_name, ) exec_called = False def fake_execvpe(file: str, args: list[str], env: dict[str, str]) -> None: nonlocal exec_called exec_called = True monkeypatch.setattr(command_router.os, "execvpe", fake_execvpe) exit_code = router.main(["cld", "chat"]) captured = capsys.readouterr() assert exit_code == 1 assert exec_called is False assert "cld: no healthy account available: avi=unknown-account" in captured.err assert "cdx:" not in captured.err def test_cli_usage_summary_uses_zero_minutes_for_expired_reset() -> None: snapshot = AccountSnapshot( HealthStatus.OK, 90, 80, primary_reset_at=1_700_000_000.0, secondary_reset_at=1_699_999_999.0, ) assert command_router._format_cli_usage_summary(snapshot, now=1_700_000_000.0) == ( "5h: 90% (0m left), 7d: 80% (0m left)" ) def test_codex_router_explicit_override_resolves_via_legacy_candidate_dir( home: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: legacy_home = home / ".codex-tray" / "accounts" / "legacyacct" / "CODEX_HOME" legacy_home.mkdir(parents=True) (home / ".codex-tray" / "accounts.json").write_text( json.dumps({"accounts": [{"slug": "legacyacct", "alias": "Legacy"}]}), encoding="utf-8", ) router = command_router.CommandRouter( command_router.CodexAdapter(), detect_project_name=command_router.detect_project_name, ) monkeypatch.setattr( command_router.CodexAdapter, "sync_before_exec", lambda self, base_dir: None, ) exit_code = router.main(["cdx", "--account", "legacyacct", "chat"]) assert exit_code == remote_dispatch.EXIT_UNCONTAINABLE def test_codex_router_explicit_override_resolves_labels_to_slug_chain( home: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: tray = home / ".local/state/overdeck/systray/runtime" for slug in ("roy", "work"): (tray / "accounts" / slug / "CODEX_HOME").mkdir(parents=True) (tray / "accounts.json").write_text( json.dumps( {"accounts": [{"slug": "roy", "alias": "Personal"}, {"slug": "work", "alias": "Office"}]} ), encoding="utf-8", ) router = command_router.CommandRouter(command_router.CodexAdapter()) resolved = router._resolve_account("personal,OFFICE") assert resolved == (tray, "roy", tray / "accounts" / "roy" / "CODEX_HOME") assert router._last_route.chain == ("roy", "work") def test_codex_router_explicit_override_resolves_dynamic_after_labels( home: Path, ) -> None: tray = home / ".local/state/overdeck/systray/runtime" for slug in ("roy", "work"): (tray / "accounts" / slug / "CODEX_HOME").mkdir(parents=True) (tray / "accounts.json").write_text( json.dumps( {"accounts": [{"slug": "roy", "alias": "Personal"}, {"slug": "work", "alias": "Office"}]} ), encoding="utf-8", ) (tray / "default_slug").write_text("work\n", encoding="utf-8") router = command_router.CommandRouter(command_router.CodexAdapter()) resolved = router._resolve_account("Personal,dynamic") assert resolved is not None assert router._last_route.chain == ("roy", "work") def test_codex_adapter_credential_builds_expected_identity(home: Path) -> None: adapter = command_router.CodexAdapter() account_home = home / ".local/state/overdeck/systray/runtime/accounts/acct/CODEX_HOME" account_home.mkdir(parents=True) credential = adapter.credential({"CODEX_HOME": str(account_home)}) assert credential == remote_dispatch.Credential( runtime="codex", slug="acct", path=account_home / "auth.json", ) def test_claude_adapter_credential_builds_expected_identity(home: Path) -> None: adapter = command_router.ClaudeAdapter() account_home = ( home / ".local/state/overdeck/systray/runtime/claude-accounts/acct/CLAUDE_HOME" ) account_home.mkdir(parents=True) credential = adapter.credential({"CLAUDE_CONFIG_DIR": str(account_home)}) assert credential == remote_dispatch.Credential( runtime="claude", slug="acct", path=account_home / ".credentials.json", ) def test_codex_route_canonical_maps_labels_to_ids_and_keeps_dynamic_literal( home: Path, capsys: pytest.CaptureFixture[str], ) -> None: tray = home / ".local/state/overdeck/systray/runtime" for slug in ("roy", "work"): (tray / "accounts" / slug / "CODEX_HOME").mkdir(parents=True) (tray / "accounts.json").write_text( json.dumps( {"accounts": [{"slug": "roy", "alias": "Personal"}, {"slug": "work", "alias": "Office"}]} ), encoding="utf-8", ) router = command_router.CommandRouter(command_router.CodexAdapter()) exit_code = router.main(["cdx", "route", "--canonical", "--profile=dynamic,OFFICE"]) captured = capsys.readouterr() assert exit_code == 0 assert json.loads(captured.out) == {"profile": "dynamic,work"} def test_codex_route_canonical_fails_closed_on_unknown_token( home: Path, capsys: pytest.CaptureFixture[str], ) -> None: tray = home / ".local/state/overdeck/systray/runtime" (tray / "accounts" / "roy" / "CODEX_HOME").mkdir(parents=True) (tray / "accounts.json").write_text( json.dumps({"accounts": [{"slug": "roy", "alias": "Personal"}]}), encoding="utf-8", ) router = command_router.CommandRouter(command_router.CodexAdapter()) exit_code = router.main(["cdx", "route", "--canonical", "--profile=ghost"]) captured = capsys.readouterr() assert exit_code == 1 assert "unknown account: ghost" in captured.err assert captured.out.strip() == "" def test_is_exec_invocation_detects_exec_after_flags() -> None: assert command_router.is_exec_invocation(["exec", "-m", "gpt-5.4", "prompt"]) assert command_router.is_exec_invocation(["exec"]) assert command_router.is_exec_invocation(["--ask-for-approval", "never", "exec", "prompt"]) assert command_router.is_exec_invocation(["--profile=work", "exec", "prompt"]) assert not command_router.is_exec_invocation(["chat"]) assert not command_router.is_exec_invocation([]) assert not command_router.is_exec_invocation(["--status"]) assert not command_router.is_exec_invocation(["--ask-for-approval", "exec"]) def test_codex_preserves_global_options_before_exec() -> None: assert command_router.normalize_codex_exec_flags( ["--ask-for-approval", "never", "exec", "-m", "gpt-5.4", "prompt"] ) == [ "--ask-for-approval", "never", "exec", "--sandbox", "danger-full-access", "--skip-git-repo-check", "-m", "gpt-5.4", "prompt", ] def test_codex_injects_sandbox_and_git_check_for_exec() -> None: assert command_router.normalize_codex_exec_flags( ["exec", "-m", "gpt-5.4", "prompt"] ) == [ "exec", "--sandbox", "danger-full-access", "--skip-git-repo-check", "-m", "gpt-5.4", "prompt", ] @pytest.mark.parametrize( "restrictive", [ ["--sandbox", "read-only"], ["-s", "workspace-write"], ["--sandbox=read-only"], ["-s=workspace-write"], ["-sread-only"], ], ) def test_codex_forces_danger_full_access_over_restrictive_flag( restrictive: list[str], ) -> None: assert command_router.normalize_codex_exec_flags( ["exec", *restrictive, "prompt"] ) == ["exec", "--sandbox", "danger-full-access", "--skip-git-repo-check", "prompt"] def test_codex_preserves_bypass_flag_and_does_not_add_sandbox() -> None: assert command_router.normalize_codex_exec_flags( ["exec", "--dangerously-bypass-approvals-and-sandbox", "prompt"] ) == [ "exec", "--skip-git-repo-check", "--dangerously-bypass-approvals-and-sandbox", "prompt", ] def test_codex_does_not_duplicate_skip_git_repo_check() -> None: assert command_router.normalize_codex_exec_flags( ["exec", "--skip-git-repo-check", "-m", "gpt-5.4", "prompt"] ) == [ "exec", "--sandbox", "danger-full-access", "--skip-git-repo-check", "-m", "gpt-5.4", "prompt", ] def test_codex_default_does_not_change_claude_arguments() -> None: assert command_router.normalize_codex_exec_flags(["chat"]) == ["chat"] @pytest.mark.parametrize( ("alias", "full_model"), [ ("sol", "gpt-5.6-sol"), ("terra", "gpt-5.6-terra"), ("luna", "gpt-5.6-luna"), ], ) def test_codex_resolves_aliases_in_every_model_flag_form( alias: str, full_model: str ) -> None: cases = [ (["exec", "-m", alias, "prompt"], ["exec", "-m", full_model, "prompt"]), ( ["exec", "--model", alias, "prompt"], ["exec", "--model", full_model, "prompt"], ), (["exec", f"-m={alias}", "prompt"], ["exec", f"-m={full_model}", "prompt"]), (["exec", f"-m{alias}", "prompt"], ["exec", f"-m{full_model}", "prompt"]), ] for argv, expected in cases: assert command_router.resolve_model_aliases(argv) == expected @pytest.mark.parametrize( "model", ["gpt-5.6-sol", "gpt-5.5", "codex-auto-review"] ) def test_codex_model_alias_resolution_preserves_full_names(model: str) -> None: cases = [ ["exec", "-m", model, "prompt"], ["exec", "--model", model, "prompt"], ["exec", f"-m={model}", "prompt"], ["exec", f"-m{model}", "prompt"], ] for argv in cases: assert command_router.resolve_model_aliases(argv) == argv def test_codex_rejects_unknown_bare_model_alias() -> None: with pytest.raises(ValueError) as error: command_router.resolve_model_aliases(["exec", "-m", "banana", "prompt"]) assert "banana" in str(error.value) for alias in ("sol", "terra", "luna"): assert alias in str(error.value) def test_codex_exec_without_model_does_not_inject_model_or_effort() -> None: normalized = command_router.normalize_codex_exec_flags(["exec", "prompt"]) assert "-m" not in normalized assert "--model" not in normalized assert not any("model_reasoning_effort" in argument for argument in normalized) def test_codex_sol_alias_receives_resolved_model_default_effort() -> None: assert command_router.normalize_codex_exec_flags(["exec", "-m", "sol", "prompt"]) == [ "exec", "--sandbox", "danger-full-access", "--skip-git-repo-check", "-c", "model_reasoning_effort=low", "-m", "gpt-5.6-sol", "prompt", ] def test_codex_luna_alias_accepts_extended_effort_after_early_resolution() -> None: assert command_router.normalize_codex_exec_flags( ["exec", "-m", "luna", "-c", "model_reasoning_effort=xhigh", "prompt"] ) == [ "exec", "--sandbox", "danger-full-access", "--skip-git-repo-check", "-m", "gpt-5.6-luna", "-c", "model_reasoning_effort=xhigh", "prompt", ] def test_codex_alias_resolution_stops_at_separator() -> None: argv = ["exec", "-m", "sol", "--", "--model", "banana", "sol"] assert command_router.resolve_model_aliases(argv) == [ "exec", "-m", "gpt-5.6-sol", "--", "--model", "banana", "sol", ] assert command_router.normalize_codex_exec_flags(argv)[-4:] == [ "--", "--model", "banana", "sol", ] @pytest.mark.parametrize( ("model", "default_effort"), [ ("gpt-5.6-sol", "low"), ("gpt-5.6-terra", "medium"), ("gpt-5.6-luna", "high"), ], ) def test_codex_injects_known_model_default_effort( model: str, default_effort: str, ) -> None: assert command_router.normalize_codex_exec_flags( ["exec", "--model", model, "prompt"] ) == [ "exec", "--sandbox", "danger-full-access", "--skip-git-repo-check", "-c", f"model_reasoning_effort={default_effort}", "--model", model, "prompt", ] @pytest.mark.parametrize("model", ["gpt-5.6-sol", "gpt-5.6-terra", "gpt-5.6-luna"]) @pytest.mark.parametrize( ("effort", "canonical"), [("low", "low"), ("med", "medium"), ("medium", "medium"), ("high", "high")], ) def test_codex_canonicalizes_standard_effort_values( model: str, effort: str, canonical: str, ) -> None: assert command_router.normalize_codex_exec_flags( ["exec", "-m", model, "-c", f"model_reasoning_effort={effort}", "prompt"] ) == [ "exec", "--sandbox", "danger-full-access", "--skip-git-repo-check", "-m", model, "-c", f"model_reasoning_effort={canonical}", "prompt", ] @pytest.mark.parametrize("effort", ["xhigh", "max"]) def test_codex_luna_accepts_extended_effort(effort: str) -> None: assert command_router.normalize_codex_exec_flags( ["exec", "-m", "gpt-5.6-luna", "-c", f"model_reasoning_effort={effort}", "prompt"] ) == [ "exec", "--sandbox", "danger-full-access", "--skip-git-repo-check", "-m", "gpt-5.6-luna", "-c", f"model_reasoning_effort={effort}", "prompt", ] @pytest.mark.parametrize("effort", ["xhigh", "max"]) def test_codex_rejects_extended_effort_without_model(effort: str) -> None: with pytest.raises( ValueError, match=rf"^unsupported reasoning effort '{effort}' for model ''$", ): command_router.normalize_codex_exec_flags( ["exec", "-c", f"model_reasoning_effort={effort}", "prompt"] ) @pytest.mark.parametrize("model", ["gpt-5.6-sol", "gpt-5.6-terra", "gpt-5.5"]) @pytest.mark.parametrize("effort", ["xhigh", "max"]) def test_codex_rejects_extended_effort_for_non_luna_models( model: str, effort: str, ) -> None: with pytest.raises( ValueError, match=rf"^unsupported reasoning effort '{effort}' for model '{model}'$", ): command_router.normalize_codex_exec_flags( ["exec", "-m", model, "-c", f"model_reasoning_effort={effort}", "prompt"] ) @pytest.mark.parametrize("model", [None, "gpt-5.6-sol", "gpt-5.5"]) def test_codex_rejects_invalid_effort_for_any_effective_model(model: str | None) -> None: argv = ["exec"] if model is not None: argv += ["-m", model] argv += ["-c", "model_reasoning_effort=minimal", "prompt"] model_name = model or "" with pytest.raises( ValueError, match=rf"^invalid reasoning effort 'minimal' for model '{model_name}'$", ): command_router.normalize_codex_exec_flags(argv) @pytest.mark.parametrize( "model_flag", [ ["-m", "gpt-5.6-sol"], ["--model", "gpt-5.6-sol"], ["-m=gpt-5.6-sol"], ["--model=gpt-5.6-sol"], ], ) def test_codex_supports_all_model_flag_spellings(model_flag: list[str]) -> None: assert command_router.normalize_codex_exec_flags(["exec", *model_flag, "prompt"]) == [ "exec", "--sandbox", "danger-full-access", "--skip-git-repo-check", "-c", "model_reasoning_effort=low", *model_flag, "prompt", ] @pytest.mark.parametrize( ("config_flag", "normalized_config_flag"), [ (["-c", "model_reasoning_effort=med"], ["-c", "model_reasoning_effort=medium"]), ( ["--config", "model_reasoning_effort=med"], ["--config", "model_reasoning_effort=medium"], ), (["-c=model_reasoning_effort=med"], ["-c=model_reasoning_effort=medium"]), ( ["--config=model_reasoning_effort=med"], ["--config=model_reasoning_effort=medium"], ), ], ) def test_codex_supports_all_effort_config_flag_spellings( config_flag: list[str], normalized_config_flag: list[str], ) -> None: assert command_router.normalize_codex_exec_flags( ["exec", "-m", "gpt-5.6-terra", *config_flag, "prompt"] ) == [ "exec", "--sandbox", "danger-full-access", "--skip-git-repo-check", "-m", "gpt-5.6-terra", *normalized_config_flag, "prompt", ] def test_codex_preserves_unrelated_config_entries_in_order() -> None: assert command_router.normalize_codex_exec_flags( [ "exec", "-c", "first=1", "--config=second=two", "--model", "gpt-5.6-luna", "-c", "model_reasoning_effort=med", "--config", "third=3", "prompt", ] ) == [ "exec", "--sandbox", "danger-full-access", "--skip-git-repo-check", "-c", "first=1", "--config=second=two", "--model", "gpt-5.6-luna", "-c", "model_reasoning_effort=medium", "--config", "third=3", "prompt", ] def test_codex_repeated_model_and_effort_flags_use_final_occurrences() -> None: assert command_router.normalize_codex_exec_flags( [ "exec", "-m", "gpt-5.6-sol", "-m=gpt-5.6-luna", "-c", "model_reasoning_effort=minimal", "--config=model_reasoning_effort=max", "prompt", ] ) == [ "exec", "--sandbox", "danger-full-access", "--skip-git-repo-check", "-m", "gpt-5.6-sol", "-m=gpt-5.6-luna", "--config=model_reasoning_effort=max", "prompt", ] def test_codex_repeated_model_flags_use_final_model_for_defaults() -> None: assert command_router.normalize_codex_exec_flags( ["exec", "-m", "gpt-5.5", "--model=gpt-5.6-sol", "prompt"] ) == [ "exec", "--sandbox", "danger-full-access", "--skip-git-repo-check", "-c", "model_reasoning_effort=low", "-m", "gpt-5.5", "--model=gpt-5.6-sol", "prompt", ] @pytest.mark.parametrize("flag", ["-m", "--model", "-c", "--config"]) def test_codex_rejects_missing_separate_flag_value(flag: str) -> None: with pytest.raises(ValueError, match=rf"^{re.escape(flag)} requires a value$"): command_router.normalize_codex_exec_flags(["exec", flag]) @pytest.mark.parametrize("flag", ["-m=", "--model=", "-c=", "--config="]) def test_codex_rejects_missing_inline_flag_value(flag: str) -> None: with pytest.raises(ValueError, match=rf"^{re.escape(flag.split('=')[0])} requires a value$"): command_router.normalize_codex_exec_flags(["exec", flag]) def test_codex_supports_attached_short_model_and_config_flags() -> None: assert command_router.normalize_codex_exec_flags( ["exec", "-mgpt-5.6-luna", "-cmodel_reasoning_effort=med", "prompt"] ) == [ "exec", "--sandbox", "danger-full-access", "--skip-git-repo-check", "-mgpt-5.6-luna", "-c=model_reasoning_effort=medium", "prompt", ] def test_codex_attached_short_model_receives_default_effort() -> None: assert command_router.normalize_codex_exec_flags( ["exec", "-mgpt-5.6-sol", "prompt"] ) == [ "exec", "--sandbox", "danger-full-access", "--skip-git-repo-check", "-c", "model_reasoning_effort=low", "-mgpt-5.6-sol", "prompt", ] def test_codex_attached_short_flags_reject_non_luna_extended_effort() -> None: with pytest.raises( ValueError, match="^unsupported reasoning effort 'max' for model 'gpt-5.6-sol'$", ): command_router.normalize_codex_exec_flags( ["exec", "-mgpt-5.6-sol", "-cmodel_reasoning_effort=max", "prompt"] ) def test_codex_stops_option_parsing_at_separator() -> None: assert command_router.normalize_codex_exec_flags( [ "exec", "-m", "gpt-5.6-sol", "--", "--model=gpt-5.6-luna", "-c", "--sandbox", "--dangerously-bypass-approvals-and-sandbox", ] ) == [ "exec", "--sandbox", "danger-full-access", "--skip-git-repo-check", "-c", "model_reasoning_effort=low", "-m", "gpt-5.6-sol", "--", "--model=gpt-5.6-luna", "-c", "--sandbox", "--dangerously-bypass-approvals-and-sandbox", ] def test_codex_model_after_separator_cannot_authorize_extended_effort() -> None: with pytest.raises( ValueError, match="^unsupported reasoning effort 'max' for model 'gpt-5.6-sol'$", ): command_router.normalize_codex_exec_flags( [ "exec", "-m", "gpt-5.6-sol", "-c", "model_reasoning_effort=max", "--", "--model=gpt-5.6-luna", ] ) @pytest.mark.parametrize( "config_flag", [ ["-c", "model_reasoning_effort"], ["--config", "model_reasoning_effort"], ["-c=model_reasoning_effort"], ["--config=model_reasoning_effort"], ["-cmodel_reasoning_effort"], ], ) def test_codex_rejects_effort_config_without_value(config_flag: list[str]) -> None: with pytest.raises(ValueError, match="^model_reasoning_effort requires a value$"): command_router.normalize_codex_exec_flags( ["exec", "-m", "gpt-5.6-sol", *config_flag, "prompt"] ) def test_codex_non_exec_arguments_remain_byte_for_byte_equivalent() -> None: argv = ["chat", "-m", "gpt-5.6-sol", "-c", "model_reasoning_effort=max", "prompt"] assert command_router.normalize_codex_exec_flags(argv) == argv def test_codex_router_rejects_invalid_effort_before_routing_sync_or_exec( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: router = command_router.CommandRouter(command_router.CodexAdapter()) monkeypatch.setattr( router, "_resolve_account", lambda *args, **kwargs: pytest.fail("account resolution must not run"), ) monkeypatch.setattr( command_router.CodexAdapter, "sync_before_exec", lambda *args, **kwargs: pytest.fail("synchronization must not run"), ) monkeypatch.setattr( command_router.subprocess, "run", lambda *args, **kwargs: pytest.fail("subprocess must not run"), ) monkeypatch.setattr( command_router.os, "execvpe", lambda *args, **kwargs: pytest.fail("exec must not run"), ) exit_code = router.main( [ "cdx", "exec", "--model=gpt-5.6-sol", "--config", "model_reasoning_effort=minimal", "prompt", ] ) captured = capsys.readouterr() assert exit_code == 1 assert captured.out == "" assert captured.err == "cdx: invalid reasoning effort 'minimal' for model 'gpt-5.6-sol'\n" def test_codex_router_surfaces_unknown_alias_without_traceback( monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: router = command_router.CommandRouter(command_router.CodexAdapter()) monkeypatch.setattr( router, "_resolve_account", lambda *args, **kwargs: pytest.fail("account resolution must not run"), ) exit_code = router.main(["cdx", "exec", "-m", "banana", "prompt"]) captured = capsys.readouterr() assert exit_code != 0 assert captured.out == "" assert captured.err == ( "cdx: unknown model alias 'banana' (valid aliases: sol, terra, luna)\n" ) assert "Traceback" not in captured.err def test_codex_exec_captures_output_to_logfile_and_prints_only_path( home: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: legacy_home = home / ".codex-tray" / "accounts" / "legacyacct" / "CODEX_HOME" legacy_home.mkdir(parents=True) (home / ".codex-tray" / "accounts.json").write_text( json.dumps({"accounts": [{"slug": "legacyacct", "alias": "Legacy"}]}), encoding="utf-8", ) token_payload = base64.urlsafe_b64encode( json.dumps({"email": "legacy@example.com"}).encode() ).decode().rstrip("=") (legacy_home / "auth.json").write_text( json.dumps({"tokens": {"id_token": f"header.{token_payload}.signature"}}), encoding="utf-8", ) router = command_router.CommandRouter( command_router.CodexAdapter(), detect_project_name=command_router.detect_project_name, ) monkeypatch.setattr( command_router.CodexAdapter, "sync_before_exec", lambda self, base_dir: None, ) monkeypatch.setenv("OD_CONTAINMENT_LOG", str(tmp_path / "containment.jsonl")) monkeypatch.setattr(command_router.remote_dispatch, "should_offload", lambda: True) run_call: dict[str, object] = {} open_call: dict[str, object] = {} account_home = legacy_home forwarded = [ "exec", "--sandbox", "danger-full-access", "--skip-git-repo-check", "-c", "model_reasoning_effort=low", "-m", "gpt-5.6-sol", "do the thing", ] credential = command_router.CodexAdapter().credential({"CODEX_HOME": str(account_home)}) assert credential is not None def fake_open_session( exec_name: str, session_forwarded: list[str], **kwargs: object, ) -> remote_dispatch.Session: open_call["exec_name"] = exec_name open_call["forwarded"] = session_forwarded open_call["credential"] = kwargs["credential"] session = remote_dispatch.Session( host_name="debian1", access={"host": "100.0.0.1", "user": "user"}, root=Path("/workspace/project"), rel_dir="sandbox/workspaces/project-abc", sandbox_id="project-abc", workspace="/node/home/sandbox/workspaces/project-abc", credential=credential, argv=remote_dispatch.launcher_argv( host_name="debian1", credential=credential, exec_name="codex", forwarded_argv=session_forwarded, sandbox_id="project-abc", ), ) session.pull_back = lambda **kwargs: None return session def fake_supervise( argv: list[str], env: dict[str, str], log_path: Path, **kwargs: object ) -> int: run_call["cmd"] = argv run_call["key"] = kwargs["key"] log_path.write_bytes(b"codex model output\n") return 7 monkeypatch.setattr(command_router.remote_dispatch, "open_session", fake_open_session) monkeypatch.setattr(command_router, "supervise", fake_supervise) monkeypatch.setattr( command_router.os, "execvpe", lambda *args, **kwargs: pytest.fail("execvpe should not run for containerized cdx exec"), ) exit_code = router.main( ["cdx", "--account", "legacyacct", "exec", "-m", "gpt-5.6-sol", "do the thing"] ) captured = capsys.readouterr() assert exit_code == 7 log_line = captured.out.strip() assert captured.out == f"{log_line}\n" assert "codex model output" not in captured.out log_path = Path(log_line) assert log_path.parent == home / ".local/state/overdeck/systray/runtime" / "logs" / "cdx" assert log_path.read_bytes() == b"codex model output\n" assert re.fullmatch( rf"cdx: running on debian1 \(sandbox/workspaces/project-abc\)\n" rf"cdx: codex exec finished — exit=7 in \d+\.\ds\. The run is COMPLETE; do not wait or poll for it\.\n" rf"cdx: log {re.escape(str(log_path))} \(19 B\)\n", captured.err, ) assert run_call["cmd"] == remote_dispatch.launcher_argv( host_name="debian1", credential=credential, exec_name="codex", forwarded_argv=forwarded, sandbox_id="project-abc", ) assert run_call["key"] == "cdx" assert open_call["exec_name"] == "codex" assert open_call["forwarded"] == forwarded assert open_call["credential"] == credential log_records = [ json.loads(line) for line in (tmp_path / "containment.jsonl").read_text(encoding="utf-8").splitlines() if line.strip() ] assert len(log_records) == 1 record = log_records[0] assert record["schema"] == 1 assert record["tool"] == "cdx" assert record["runtime"] == "codex" assert record["mode"] == "container" assert record["containment"] == "podman-rootless" assert record["host"] == "debian1" assert record["workspace"] == "/node/home/sandbox/workspaces/project-abc" assert record["sandboxId"] == "project-abc" assert record["account"] == "legacyacct" assert record["logPath"] == str(log_path) assert record["exitCode"] == 7 assert record["reason"] is None assert record["durationSec"] >= 0 serialized = json.dumps(record) assert token_payload not in serialized assert str(account_home / "auth.json") not in serialized assert captured.out == f"{log_path}\n" @pytest.mark.parametrize( "case", [ "cdx_exec_container", "cdx_exec_in_container", "cdx_non_exec", "cld_any", "cdx_exec_open_session_fails", "cdx_exec_launcher_fails", ], ids=[ "cdx_exec_container", "cdx_exec_in_container", "cdx_non_exec", "cld_any", "cdx_exec_open_session_fails", "cdx_exec_launcher_fails", ], ) def test_every_agent_start_path_is_contained_or_aborts( case: str, home: Path, tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: legacy_home = home / ".codex-tray" / "accounts" / "legacyacct" / "CODEX_HOME" legacy_home.mkdir(parents=True) (legacy_home / "auth.json").write_text("token", encoding="utf-8") claude_home = home / ".local/state/overdeck/systray/runtime" / "claude-accounts" / "avi" / "CLAUDE_HOME" claude_home.mkdir(parents=True) monkeypatch.setenv("OD_CONTAINMENT_LOG", str(tmp_path / f"{case}.jsonl")) def execvpe_guard(*args: object, **kwargs: object) -> None: if not command_router.remote_dispatch.in_container(): pytest.fail("execvpe reached while not in container") return None monkeypatch.setattr(command_router.os, "execvpe", execvpe_guard) if case == "cld_any": monkeypatch.setattr( command_router.subprocess, "run", lambda *args, **kwargs: pytest.fail("vendor preflight process"), ) if case == "cdx_exec_in_container": monkeypatch.setattr(command_router.remote_dispatch, "in_container", lambda: True) else: monkeypatch.setattr(command_router.remote_dispatch, "in_container", lambda: False) counts: dict[str, int] = {"open_session": 0, "supervise": 0} def fake_open_session( exec_name: str, session_forwarded: list[str], **kwargs: object, ) -> remote_dispatch.Session: counts["open_session"] += 1 credential = kwargs["credential"] assert credential is not None session = remote_dispatch.Session( host_name="debian1", access={"host": "100.0.0.1", "user": "user"}, root=Path("/workspace/project"), rel_dir="sandbox/workspaces/project-abc", sandbox_id="project-abc", workspace="/node/home/sandbox/workspaces/project-abc", credential=credential, argv=remote_dispatch.launcher_argv( host_name="debian1", credential=credential, exec_name="codex", forwarded_argv=session_forwarded, sandbox_id="project-abc", ), ) session.pull_back = lambda **kwargs: None return session def fake_open_session_forbidden(*args: object, **kwargs: object) -> remote_dispatch.Session: counts["open_session"] += 1 raise AssertionError("open_session should not be called") supervise_rc = remote_dispatch.RC_GIT_MOUNT if case == "cdx_exec_launcher_fails" else 0 def fake_supervise( argv: list[str], env: dict[str, str], log_path: Path, **kwargs: object ) -> int: counts["supervise"] += 1 log_path.write_bytes(b"hi\n") return supervise_rc expected_stdout_is_log_path = case in {"cdx_exec_container", "cdx_exec_in_container", "cdx_exec_launcher_fails"} if case == "cdx_exec_container": monkeypatch.setattr(command_router.remote_dispatch, "should_offload", lambda: True) monkeypatch.setattr(command_router.remote_dispatch, "open_session", fake_open_session) expected_mode = "container" expected_exit = 0 elif case == "cdx_exec_in_container": monkeypatch.setattr(command_router.remote_dispatch, "should_offload", lambda: False) monkeypatch.setattr(command_router.remote_dispatch, "open_session", fake_open_session_forbidden) expected_mode = "in-container" expected_exit = 0 elif case == "cdx_exec_launcher_fails": monkeypatch.setattr(command_router.remote_dispatch, "should_offload", lambda: True) monkeypatch.setattr(command_router.remote_dispatch, "open_session", fake_open_session) expected_mode = "aborted" expected_exit = remote_dispatch.RC_GIT_MOUNT elif case == "cdx_exec_open_session_fails": monkeypatch.setattr(command_router.remote_dispatch, "should_offload", lambda: True) def failing_open_session( exec_name: str, session_forwarded: list[str], **kwargs: object, ) -> remote_dispatch.Session: counts["open_session"] += 1 raise remote_dispatch.OffloadUnavailable("node unavailable", code=remote_dispatch.EXIT_MIRROR) monkeypatch.setattr(command_router.remote_dispatch, "open_session", failing_open_session) expected_mode = "aborted" expected_exit = remote_dispatch.EXIT_MIRROR elif case == "cdx_non_exec": monkeypatch.setattr(command_router.remote_dispatch, "should_offload", lambda: True) monkeypatch.setattr(command_router.remote_dispatch, "open_session", fake_open_session_forbidden) expected_mode = "aborted" expected_exit = remote_dispatch.EXIT_UNCONTAINABLE else: monkeypatch.setattr(command_router.remote_dispatch, "should_offload", lambda: True) monkeypatch.setattr(command_router.remote_dispatch, "open_session", fake_open_session_forbidden) expected_mode = "aborted" expected_exit = remote_dispatch.EXIT_UNCONTAINABLE monkeypatch.setattr(command_router, "supervise", fake_supervise) monkeypatch.setattr( command_router.CodexAdapter, "sync_before_exec", lambda self, base_dir: None, ) monkeypatch.setattr(command_router.ClaudeAdapter, "sync_before_exec", lambda self, base_dir: None) cdx_router = command_router.CommandRouter( command_router.CodexAdapter(), detect_project_name=command_router.detect_project_name, ) cld_router = command_router.CommandRouter( command_router.ClaudeAdapter(), detect_project_name=command_router.detect_project_name, ) if case.startswith("cdx_exec"): exit_code = cdx_router.main( ["cdx", "--account", "legacyacct", "exec", "-m", "gpt-5.6-sol", "hi"] ) elif case == "cdx_non_exec": exit_code = cdx_router.main(["cdx", "--account", "legacyacct", "login", "status"]) else: exit_code = cld_router.main(["cld", "--account", "avi", "chat"]) if case in {"cdx_exec_container", "cdx_exec_launcher_fails"}: assert counts["open_session"] == 1 assert counts["supervise"] == 1 elif case == "cdx_exec_in_container": assert counts["open_session"] == 0 assert counts["supervise"] == 1 elif case == "cdx_exec_open_session_fails": assert counts["open_session"] == 1 assert counts["supervise"] == 0 else: assert counts["open_session"] == 0 assert counts["supervise"] == 0 captured = capsys.readouterr() assert exit_code == expected_exit if expected_stdout_is_log_path: assert captured.out == f"{captured.out.strip()}\n" assert captured.out.strip().endswith(".log") else: assert captured.out == "" records = [ json.loads(line) for line in (tmp_path / f"{case}.jsonl").read_text(encoding="utf-8").splitlines() if line.strip() ] assert len(records) == 1 record = records[0] assert record["mode"] == expected_mode if case.startswith("cdx"): assert record["tool"] == "cdx" assert record["runtime"] == "codex" else: assert record["tool"] == "cld" assert record["runtime"] == "claude" assert record["exitCode"] == expected_exit if expected_mode == "container": assert record["host"] == "debian1" assert record["workspace"] == "/node/home/sandbox/workspaces/project-abc" assert record["sandboxId"] == "project-abc" assert record["containment"] == "podman-rootless" assert record["account"] == "legacyacct" assert record["reason"] is None elif expected_mode == "in-container": assert record["host"] is None assert record["workspace"] is None assert record["sandboxId"] is None assert record["containment"] == "podman-rootless" assert record["account"] == "legacyacct" assert record["reason"] is None else: assert record["host"] is None assert record["workspace"] is None assert record["sandboxId"] is None assert record["containment"] == "none" assert record["account"] == ("legacyacct" if case.startswith("cdx") else "avi") expected_reasons = { "cdx_exec_open_session_fails": "node unavailable", "cdx_exec_launcher_fails": f"debian1: the sandbox launcher started no container (rc {remote_dispatch.RC_GIT_MOUNT})", } assert record["reason"] == expected_reasons.get( case, "adapter captures no log for these arguments" ) if case in {"cdx_exec_open_session_fails", "cdx_exec_launcher_fails"}: assert f"cdx: containment unavailable — {record['reason']}" in captured.err def test_run_captured_releases_the_placement_claim_when_the_run_raises( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: claims = tmp_path / "claims" session = remote_dispatch.Session( host_name="debian1", access={"host": "100.0.0.1", "user": "user"}, root=tmp_path, rel_dir="cdx-offload/x", argv=["ssh", "user@100.0.0.1"], claim=remote_dispatch.acquire_claim("debian1", claims), ) assert len(list(claims.glob("*.claim"))) == 1 def boom(*args: object, **kwargs: object) -> int: raise OSError("the node went away mid-run") monkeypatch.setattr(session, "pull_back", lambda **kwargs: None) monkeypatch.setattr(command_router.remote_dispatch, "should_offload", lambda: True) monkeypatch.setattr( command_router.remote_dispatch, "open_session", lambda *a, **kw: session ) monkeypatch.setattr(command_router, "supervise", boom) router = command_router.CommandRouter( command_router.CodexAdapter(), detect_project_name=command_router.detect_project_name, ) exit_code = router._run_captured(["exec", "hi"], {}, tmp_path / "run.log", None, time.monotonic()) assert exit_code == 1 assert list(claims.glob("*.claim")) == [] assert "the node went away mid-run" in capsys.readouterr().err def test_run_captured_passes_adapter_credential_into_open_session( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: account_home = tmp_path / "accounts" / "acct" / "CODEX_HOME" account_home.mkdir(parents=True) (account_home / "auth.json").write_text("token", encoding="utf-8") expected = remote_dispatch.Credential( runtime="codex", slug="acct", path=account_home / "auth.json", ) captured: dict[str, object] = {} def fake_open_session( exec_name: str, forwarded_argv: list[str], **kwargs: object, ) -> remote_dispatch.Session: captured["credential"] = kwargs["credential"] session = remote_dispatch.Session( host_name="debian1", access={"host": "100.0.0.1", "user": "user"}, root=tmp_path, rel_dir="cdx-offload/x", argv=["agent-sandbox"], ) session.pull_back = lambda **kwargs: None return session monkeypatch.setattr(command_router.remote_dispatch, "should_offload", lambda: True) monkeypatch.setattr(command_router.remote_dispatch, "open_session", fake_open_session) monkeypatch.setattr(command_router, "supervise", lambda argv, env, log_path, **kw: 0) router = command_router.CommandRouter( command_router.CodexAdapter(), detect_project_name=command_router.detect_project_name, ) env = {"CODEX_HOME": str(account_home)} resolved = router.adapter.credential(env) assert resolved == expected router._run_captured(["exec", "hi"], env, tmp_path / "run.log", resolved, time.monotonic()) assert captured["credential"] == expected @pytest.mark.parametrize("entrypoint", ["_run_captured", "main"]) @pytest.mark.parametrize("escape_env", [False, True]) def test_run_or_main_rejects_offload_unavailable( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], entrypoint: str, escape_env: bool, ) -> None: if escape_env: for env_name in ( "CDX_NO_OFFLOAD", "OD_REMOTE_EXEC", "HARNESS_SEAT_CONTAINER", "CDX_OFFLOAD_STRICT", ): monkeypatch.setenv(env_name, "1" if env_name != "CDX_OFFLOAD_STRICT" else "0") monkeypatch.setattr(command_router.remote_dispatch, "in_container", lambda: False) def raise_unavailable(*_args: object, **_kwargs: object) -> None: raise remote_dispatch.OffloadUnavailable( "no usable access door", code=remote_dispatch.EXIT_MIRROR ) monkeypatch.setattr( command_router.remote_dispatch, "open_session", raise_unavailable, ) supervise_called = False def fake_supervise( argv: list[str], env: dict[str, str], log_path: Path, **kwargs: object ) -> int: nonlocal supervise_called supervise_called = True return 0 monkeypatch.setattr(command_router, "supervise", fake_supervise) router = command_router.CommandRouter( command_router.CodexAdapter(), detect_project_name=command_router.detect_project_name, ) log_path = tmp_path / "run.log" if entrypoint == "_run_captured": exit_code = router._run_captured( ["exec", "hello"], {**os.environ, "CODEX_HOME": str(tmp_path / "accounts" / "acct" / "CODEX_HOME")}, log_path, None, time.monotonic(), ) else: account_home = tmp_path / "accounts" / "acct" / "CODEX_HOME" account_home.mkdir(parents=True) monkeypatch.setattr( command_router.CodexAdapter, "sync_before_exec", lambda self, base_dir: None, ) monkeypatch.setattr( router, "_resolve_account", lambda *args, **kwargs: (tmp_path, "acct", account_home), ) monkeypatch.setattr( command_router.CodexAdapter, "capture_log_path", lambda self, forwarded_argv: log_path, ) exit_code = router.main(["cdx", "exec", "hello"]) assert exit_code == remote_dispatch.EXIT_MIRROR assert not supervise_called assert "cdx: containment unavailable — no usable access door" in capsys.readouterr().err def test_cdx_main_with_no_containerized_log_path_returns_uncontainable( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: router = command_router.CommandRouter( command_router.CodexAdapter(), detect_project_name=command_router.detect_project_name, ) account_home = tmp_path / "accounts" / "acct" / "CODEX_HOME" account_home.mkdir(parents=True) monkeypatch.setattr( router, "_resolve_account", lambda *args, **kwargs: (tmp_path, "acct", account_home), ) monkeypatch.setattr( command_router.remote_dispatch, "in_container", lambda: False, ) monkeypatch.setattr( command_router.CodexAdapter, "capture_log_path", lambda self, forwarded_argv: None, ) monkeypatch.setattr( command_router.remote_dispatch, "open_session", lambda *args, **kwargs: pytest.fail("open_session should not run when capture_log_path is None"), ) exec_called = False def fake_execvpe(file: str, args: list[str], env: dict[str, str]) -> None: nonlocal exec_called exec_called = True monkeypatch.setattr(command_router.os, "execvpe", fake_execvpe) exit_code = router.main(["cdx", "chat"]) captured = capsys.readouterr() assert exit_code == remote_dispatch.EXIT_UNCONTAINABLE assert exec_called is False assert captured.err.strip() == ( "cdx: no containerized path for this invocation (adapter captures no log for these arguments)" ) def test_codex_router_explicit_override_missing_everywhere_reports_unknown( home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: router = command_router.CommandRouter( command_router.CodexAdapter(), detect_project_name=command_router.detect_project_name, ) exit_code = router.main(["cdx", "--account", "nope", "chat"]) captured = capsys.readouterr() assert exit_code == 1 assert captured.err.strip() == "cdx: unknown account 'nope'" def test_codex_router_reports_clean_error_for_missing_rules_file( home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: (home / ".local/state/overdeck/systray/runtime").mkdir(parents=True) monkeypatch.setattr(command_router, "detect_project_name", lambda cwd=None: "zync.is") router = command_router.CommandRouter( command_router.CodexAdapter(), detect_project_name=command_router.detect_project_name, ) exit_code = router.main(["cdx", "chat"]) captured = capsys.readouterr() assert exit_code == 1 assert captured.err.strip().startswith("cdx: failed to load routing rules:") def test_codex_router_reports_clean_error_for_malformed_rules_json( home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: tray_dir = home / ".local/state/overdeck/systray/runtime" tray_dir.mkdir(parents=True) (tray_dir / "routing_rules.json").write_text("not-json", encoding="utf-8") monkeypatch.setattr(command_router, "detect_project_name", lambda cwd=None: "zync.is") router = command_router.CommandRouter( command_router.CodexAdapter(), detect_project_name=command_router.detect_project_name, ) exit_code = router.main(["cdx", "chat"]) captured = capsys.readouterr() assert exit_code == 1 assert captured.err.strip().startswith("cdx: failed to load routing rules:") def test_codex_router_reports_clean_error_for_invalid_rules_shape( home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: write_rules( home, { "filename": "routing_rules.json", "rules": { "projects": {}, "default": "", "fallback_chain": [], "fallback_trigger": "broken_only", "quota_exhausted_threshold_pct": 100, }, }, ) monkeypatch.setattr(command_router, "detect_project_name", lambda cwd=None: "zync.is") router = command_router.CommandRouter( command_router.CodexAdapter(), detect_project_name=command_router.detect_project_name, ) exit_code = router.main(["cdx", "chat"]) captured = capsys.readouterr() assert exit_code == 1 assert captured.err.strip().startswith("cdx: failed to load routing rules:") def test_claude_router_execs_with_claude_config_dir( home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: write_rules( home, { "filename": "claude_routing_rules.json", "rules": { "projects": {}, "default": "avi", "fallback_chain": ["roy"], "fallback_trigger": "broken_or_quota_exhausted", "quota_exhausted_threshold_pct": 100, }, }, ) claude_home = home / ".local/state/overdeck/systray/runtime" / "claude-accounts" / "avi" / "CLAUDE_HOME" claude_home.mkdir(parents=True) write_health( home, { "avi": {"status": "ok", "primary_used_pct": 10, "secondary_used_pct": 20}, }, "claude_health_cache.json", ) monkeypatch.setattr(command_router, "detect_project_name", lambda cwd=None: "unknown") monkeypatch.setattr(command_router.ClaudeAdapter, "verify_env_support", lambda self, home: True) router = command_router.CommandRouter( command_router.ClaudeAdapter(), detect_project_name=command_router.detect_project_name, ) monkeypatch.setattr(command_router.os, "execvpe", lambda *args, **kwargs: pytest.fail("execvpe should not run")) exit_code = router.main(["cld", "chat"]) captured = capsys.readouterr() assert exit_code == remote_dispatch.EXIT_UNCONTAINABLE assert captured.err.strip() == ( "cld: no containerized path for this invocation (adapter captures no log for these arguments)" ) def test_router_keeps_the_notification_opt_in_out_of_the_child_env( home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: write_rules( home, { "filename": "claude_routing_rules.json", "rules": { "projects": {}, "default": "avi", "fallback_chain": ["roy"], "fallback_trigger": "broken_or_quota_exhausted", "quota_exhausted_threshold_pct": 100, }, }, ) (home / ".local/state/overdeck/systray/runtime" / "claude-accounts" / "avi" / "CLAUDE_HOME").mkdir(parents=True) write_health( home, {"avi": {"status": "ok", "primary_used_pct": 10, "secondary_used_pct": 20}}, "claude_health_cache.json", ) monkeypatch.setenv("STALL_GUARD_NOTIFY", "1") monkeypatch.setattr(command_router, "detect_project_name", lambda cwd=None: "unknown") monkeypatch.setattr(command_router.ClaudeAdapter, "verify_env_support", lambda self, home: True) router = command_router.CommandRouter( command_router.ClaudeAdapter(), detect_project_name=command_router.detect_project_name, ) monkeypatch.setattr(command_router.os, "execvpe", lambda *args, **kwargs: pytest.fail("execvpe should not run")) exit_code = router.main(["cld", "chat"]) captured = capsys.readouterr() assert exit_code == remote_dispatch.EXIT_UNCONTAINABLE assert os.environ["STALL_GUARD_NOTIFY"] == "1" assert captured.err.strip() == ( "cld: no containerized path for this invocation (adapter captures no log for these arguments)" ) def test_codex_router_ignores_conflicting_claude_health_cache( home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: write_rules( home, { "filename": "routing_rules.json", "rules": { "projects": {}, "default": "shared", "fallback_chain": [], "fallback_trigger": "broken_or_quota_exhausted", "quota_exhausted_threshold_pct": 100, }, }, ) codex_home = home / ".local/state/overdeck/systray/runtime" / "accounts" / "shared" / "CODEX_HOME" codex_home.mkdir(parents=True) (home / ".local/state/overdeck/systray/runtime" / "claude-accounts" / "shared" / "CLAUDE_HOME").mkdir(parents=True) write_health( home, { "shared": {"status": "ok", "primary_used_pct": 10, "secondary_used_pct": 20}, }, "health_cache.json", ) write_health( home, { "shared": {"status": "broken", "primary_used_pct": None, "secondary_used_pct": None}, }, "claude_health_cache.json", ) monkeypatch.setattr(command_router, "detect_project_name", lambda cwd=None: "unknown") monkeypatch.setattr( command_router.CodexAdapter, "sync_before_exec", lambda self, base_dir: None, ) router = command_router.CommandRouter( command_router.CodexAdapter(), detect_project_name=command_router.detect_project_name, ) monkeypatch.setattr(command_router.os, "execvpe", lambda *args, **kwargs: pytest.fail("execvpe should not run")) exit_code = router.main(["cdx", "chat"]) captured = capsys.readouterr() assert exit_code == remote_dispatch.EXIT_UNCONTAINABLE assert captured.err.strip() == ( "cdx: no containerized path for this invocation (adapter captures no log for these arguments)" ) def test_claude_router_ignores_conflicting_codex_health_cache( home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: write_rules( home, { "filename": "claude_routing_rules.json", "rules": { "projects": {}, "default": "shared", "fallback_chain": [], "fallback_trigger": "broken_or_quota_exhausted", "quota_exhausted_threshold_pct": 100, }, }, ) (home / ".local/state/overdeck/systray/runtime" / "accounts" / "shared" / "CODEX_HOME").mkdir(parents=True) claude_home = home / ".local/state/overdeck/systray/runtime" / "claude-accounts" / "shared" / "CLAUDE_HOME" claude_home.mkdir(parents=True) write_health( home, { "shared": {"status": "broken", "primary_used_pct": None, "secondary_used_pct": None}, }, "health_cache.json", ) write_health( home, { "shared": {"status": "ok", "primary_used_pct": 10, "secondary_used_pct": 20}, }, "claude_health_cache.json", ) monkeypatch.setattr(command_router, "detect_project_name", lambda cwd=None: "unknown") monkeypatch.setattr(command_router.ClaudeAdapter, "verify_env_support", lambda self, home: True) router = command_router.CommandRouter( command_router.ClaudeAdapter(), detect_project_name=command_router.detect_project_name, ) monkeypatch.setattr(command_router.os, "execvpe", lambda *args, **kwargs: pytest.fail("execvpe should not run")) exit_code = router.main(["cld", "chat"]) captured = capsys.readouterr() assert exit_code == remote_dispatch.EXIT_UNCONTAINABLE assert captured.err.strip() == ( "cld: no containerized path for this invocation (adapter captures no log for these arguments)" ) def test_claude_router_notifies_and_uses_fallback_for_broken_default( home: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: write_rules( home, { "filename": "claude_routing_rules.json", "rules": { "version": "routing/v2", "projects": { "unknown": {"account": "avi", "fallback": ["roy"]}, }, "default": "avi", "quota_exhausted_threshold_pct": 100, }, }, ) claude_default_home = home / ".local/state/overdeck/systray/runtime" / "claude-accounts" / "avi" / "CLAUDE_HOME" claude_default_home.mkdir(parents=True) claude_fallback_home = home / ".local/state/overdeck/systray/runtime" / "claude-accounts" / "roy" / "CLAUDE_HOME" claude_fallback_home.mkdir(parents=True) write_health( home, { "avi": {"status": "broken", "primary_used_pct": None, "secondary_used_pct": None}, "roy": {"status": "ok", "primary_used_pct": 10, "secondary_used_pct": 20}, }, "claude_health_cache.json", ) monkeypatch.setattr(command_router, "detect_project_name", lambda cwd=None: "unknown") monkeypatch.setattr( command_router.ClaudeAdapter, "verify_env_support", lambda self, account_home: True, raising=False, ) router = command_router.CommandRouter( command_router.ClaudeAdapter(), detect_project_name=command_router.detect_project_name, ) popen_calls: list[list[str]] = [] def fake_popen(args: list[str]) -> object: popen_calls.append(args) return object() monkeypatch.setattr(command_router.os, "execvpe", lambda *args, **kwargs: pytest.fail("execvpe should not run")) monkeypatch.setattr(command_router.subprocess, "Popen", fake_popen) exit_code = router.main(["cld", "chat"]) assert exit_code == remote_dispatch.EXIT_UNCONTAINABLE assert popen_calls == [["notify-send", "cld", "avi unavailable, using roy instead"]] def test_claude_adapter_build_env_identifies_statusline_account(home: Path) -> None: account_home = home / ".local/state/overdeck/systray/runtime" / "claude-accounts" / "multideal" / "CLAUDE_HOME" assert command_router.ClaudeAdapter().build_env(account_home) == { "CLAUDE_CONFIG_DIR": str(account_home), "SYSTRAY_CLAUDE_ACCOUNT_HOME": str(account_home), "SYSTRAY_CLAUDE_ACCOUNT_SLUG": "multideal", } def test_claude_adapter_uses_declared_environment_contract_without_probe( home: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: account_home = home / "claude-accounts" / "avi" / "CLAUDE_HOME" monkeypatch.setattr( command_router.subprocess, "run", lambda *args, **kwargs: pytest.fail("runtime capability probe"), ) assert command_router.ClaudeAdapter().verify_env_support(account_home) is True def test_claude_router_rejects_unsupported_routing_when_probe_fails( home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: write_rules( home, { "filename": "claude_routing_rules.json", "rules": { "projects": {}, "default": "avi", "fallback_chain": ["roy"], "fallback_trigger": "broken_or_quota_exhausted", "quota_exhausted_threshold_pct": 100, }, }, ) claude_home = home / ".local/state/overdeck/systray/runtime" / "claude-accounts" / "avi" / "CLAUDE_HOME" claude_home.mkdir(parents=True) write_health( home, { "avi": {"status": "ok", "primary_used_pct": 10, "secondary_used_pct": 20}, }, "claude_health_cache.json", ) monkeypatch.setattr(command_router, "detect_project_name", lambda cwd=None: "unknown") monkeypatch.setattr( command_router.ClaudeAdapter, "verify_env_support", lambda self, account_home: False, raising=False, ) router = command_router.CommandRouter( command_router.ClaudeAdapter(), detect_project_name=command_router.detect_project_name, ) exec_called = False def fake_execvpe(file: str, args: list[str], env: dict[str, str]) -> None: nonlocal exec_called exec_called = True monkeypatch.setattr(command_router.os, "execvpe", fake_execvpe) exit_code = router.main(["cld", "chat"]) captured = capsys.readouterr() assert exit_code == 1 assert exec_called is False assert captured.err.strip() == ( f"cld: unsupported routing: claude did not honor CLAUDE_CONFIG_DIR for {claude_home}" ) def test_capture_log_path_falls_back_when_tray_logs_unwritable( home: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: tray_logs = home / ".local/state/overdeck/systray/runtime" / "logs" tray_logs.mkdir(parents=True) tray_logs.chmod(0o500) fallback_root = home / "tmpdir" fallback_root.mkdir() monkeypatch.setattr(command_router.tempfile, "gettempdir", lambda: str(fallback_root)) log_path = command_router.CodexAdapter().capture_log_path(["exec", "hi"]) assert log_path is not None assert log_path.parent == fallback_root / "cdx-logs" assert log_path.parent.is_dir() def test_capture_log_path_returns_none_when_no_directory_is_writable( home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: tray_logs = home / ".local/state/overdeck/systray/runtime" / "logs" tray_logs.mkdir(parents=True) tray_logs.chmod(0o500) fallback_root = home / "tmpdir" fallback_root.mkdir(mode=0o500) monkeypatch.setattr(command_router.tempfile, "gettempdir", lambda: str(fallback_root)) log_path = command_router.CodexAdapter().capture_log_path(["exec", "hi"]) assert log_path is None assert "no writable log directory" in capsys.readouterr().err def test_codex_router_all_capped_exits_75_with_resume_at( home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: write_rules( home, { "filename": "routing_rules.json", "rules": { "projects": {}, "default": "rafa", "fallback_chain": ["roy"], "fallback_trigger": "broken_or_quota_exhausted", "quota_exhausted_threshold_pct": 100, "account_caps": { "rafa": {"7d": 10}, "roy": {"7d": 10}, }, }, }, ) for slug in ("rafa", "roy"): (home / ".local/state/overdeck/systray/runtime" / "accounts" / slug / "CODEX_HOME").mkdir(parents=True) write_health( home, { "rafa": { "status": "ok", "primary_used_pct": 10, "secondary_used_pct": 90, "secondary_reset_at": 1786365653.0, }, "roy": { "status": "ok", "primary_used_pct": 10, "secondary_used_pct": 95, "secondary_reset_at": 1787000000.0, }, }, "health_cache.json", ) monkeypatch.setattr(command_router, "detect_project_name", lambda cwd=None: "unknown") router = command_router.CommandRouter( command_router.CodexAdapter(), detect_project_name=command_router.detect_project_name, ) exec_called = False def fake_execvpe(file: str, args: list[str], env: dict[str, str]) -> None: nonlocal exec_called exec_called = True monkeypatch.setattr(command_router.os, "execvpe", fake_execvpe) exit_code = router.main(["cdx", "chat"]) captured = capsys.readouterr() assert exit_code == 75 assert exec_called is False assert captured.err == "" assert json.loads(captured.out) == { "ok": False, "detail": "rate-limited", "reason": "all-accounts-capped", "resume_at": "2026-08-10T12:40:53Z", } def test_cld_router_all_capped_exits_75_without_resume_at_when_unknown( home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: write_rules( home, { "filename": "claude_routing_rules.json", "rules": { "projects": {}, "default": "avi", "fallback_chain": [], "fallback_trigger": "broken_or_quota_exhausted", "quota_exhausted_threshold_pct": 100, "account_caps": {"avi": {"7d": 10}}, }, }, ) (home / ".local/state/overdeck/systray/runtime" / "claude-accounts" / "avi" / "CLAUDE_HOME").mkdir(parents=True) write_health( home, { "avi": { "status": "ok", "primary_used_pct": 10, "secondary_used_pct": 90, }, }, "claude_health_cache.json", ) monkeypatch.setattr(command_router, "detect_project_name", lambda cwd=None: "unknown") router = command_router.CommandRouter( command_router.ClaudeAdapter(), detect_project_name=command_router.detect_project_name, ) exec_called = False def fake_execvpe(file: str, args: list[str], env: dict[str, str]) -> None: nonlocal exec_called exec_called = True monkeypatch.setattr(command_router.os, "execvpe", fake_execvpe) exit_code = router.main(["cld", "route"]) captured = capsys.readouterr() assert exit_code == 75 assert exec_called is False assert captured.err == "" assert json.loads(captured.out) == { "ok": False, "detail": "rate-limited", "reason": "all-accounts-capped", } assert "resume_at" not in captured.out def test_codex_router_all_quota_exhausted_stays_exit_one_not_all_capped( home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: write_rules( home, { "filename": "routing_rules.json", "rules": { "projects": {}, "default": "rafa", "fallback_chain": ["roy"], "fallback_trigger": "broken_or_quota_exhausted", "quota_exhausted_threshold_pct": 100, }, }, ) for slug in ("rafa", "roy"): (home / ".local/state/overdeck/systray/runtime" / "accounts" / slug / "CODEX_HOME").mkdir(parents=True) write_health( home, { "rafa": {"status": "ok", "primary_used_pct": 100, "secondary_used_pct": 10}, "roy": {"status": "ok", "primary_used_pct": 10, "secondary_used_pct": 100}, }, "health_cache.json", ) monkeypatch.setattr(command_router, "detect_project_name", lambda cwd=None: "unknown") router = command_router.CommandRouter( command_router.CodexAdapter(), detect_project_name=command_router.detect_project_name, ) exec_called = False def fake_execvpe(file: str, args: list[str], env: dict[str, str]) -> None: nonlocal exec_called exec_called = True monkeypatch.setattr(command_router.os, "execvpe", fake_execvpe) exit_code = router.main(["cdx", "chat"]) captured = capsys.readouterr() assert exit_code == 1 assert exec_called is False assert "all-accounts-capped" not in captured.out assert "quota-exhausted" in captured.err def test_codex_router_explicit_override_refuses_capped_account( home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: write_rules( home, { "filename": "routing_rules.json", "rules": { "projects": {}, "default": "zync2", "fallback_chain": [], "fallback_trigger": "broken_or_quota_exhausted", "quota_exhausted_threshold_pct": 100, "account_caps": {"zync2": {"7d": 10}}, }, }, ) tray = home / ".local/state/overdeck/systray/runtime" (tray / "accounts" / "zync2" / "CODEX_HOME").mkdir(parents=True) write_health( home, { "zync2": { "status": "ok", "primary_used_pct": 10, "secondary_used_pct": 91, "secondary_reset_at": 1786365653.0, }, }, "health_cache.json", ) router = command_router.CommandRouter(command_router.CodexAdapter()) from routing_resolver import AllAccountsCappedError, RoutingResolver, load_rules resolved = router._resolve_account("zync2") assert isinstance(resolved, AllAccountsCappedError) rules = load_rules(tray / "routing_rules.json") health = command_router.load_health(tray / "health_cache.json") resolver = RoutingResolver(rules, health, known_slugs={"zync2"}) assert resolver._describe_status("zync2") == "capped(7d=9% left <= 10%)" exec_called = False def fake_execvpe(file: str, args: list[str], env: dict[str, str]) -> None: nonlocal exec_called exec_called = True monkeypatch.setattr(command_router.os, "execvpe", fake_execvpe) exit_code = router.main(["cdx", "--account", "zync2", "chat"]) captured = capsys.readouterr() assert exit_code == 75 assert exec_called is False assert json.loads(captured.out) == { "ok": False, "detail": "rate-limited", "reason": "all-accounts-capped", "resume_at": "2026-08-10T12:40:53Z", } def test_codex_router_explicit_override_skips_capped_chain_entry(home: Path) -> None: write_rules( home, { "filename": "routing_rules.json", "rules": { "projects": {}, "default": "zync2", "fallback_chain": [], "fallback_trigger": "broken_or_quota_exhausted", "quota_exhausted_threshold_pct": 100, "account_caps": { "zync2": {"7d": 10}, "roy": {"7d": 10}, }, }, }, ) tray = home / ".local/state/overdeck/systray/runtime" for slug in ("zync2", "roy"): (tray / "accounts" / slug / "CODEX_HOME").mkdir(parents=True) write_health( home, { "zync2": { "status": "ok", "primary_used_pct": 10, "secondary_used_pct": 91, }, "roy": { "status": "ok", "primary_used_pct": 10, "secondary_used_pct": 10, }, }, "health_cache.json", ) router = command_router.CommandRouter(command_router.CodexAdapter()) resolved = router._resolve_account("zync2,roy") assert resolved == (tray, "roy", tray / "accounts" / "roy" / "CODEX_HOME") assert router._last_route.slug == "roy" assert router._last_route.chain == ("zync2", "roy") def test_codex_router_explicit_override_all_capped_returns_all_accounts_capped_error( home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: write_rules( home, { "filename": "routing_rules.json", "rules": { "projects": {}, "default": "zync2", "fallback_chain": [], "fallback_trigger": "broken_or_quota_exhausted", "quota_exhausted_threshold_pct": 100, "account_caps": { "zync2": {"7d": 10}, "roy": {"7d": 10}, }, }, }, ) tray = home / ".local/state/overdeck/systray/runtime" for slug in ("zync2", "roy"): (tray / "accounts" / slug / "CODEX_HOME").mkdir(parents=True) write_health( home, { "zync2": { "status": "ok", "primary_used_pct": 10, "secondary_used_pct": 91, "secondary_reset_at": 1786365653.0, }, "roy": { "status": "ok", "primary_used_pct": 10, "secondary_used_pct": 95, "secondary_reset_at": 1787000000.0, }, }, "health_cache.json", ) router = command_router.CommandRouter(command_router.CodexAdapter()) from routing_resolver import AllAccountsCappedError resolved = router._resolve_account("zync2,roy") assert isinstance(resolved, AllAccountsCappedError) assert resolved.resume_at == "2026-08-10T12:40:53Z" exec_called = False def fake_execvpe(file: str, args: list[str], env: dict[str, str]) -> None: nonlocal exec_called exec_called = True monkeypatch.setattr(command_router.os, "execvpe", fake_execvpe) exit_code = router.main(["cdx", "--account", "zync2,roy", "chat"]) captured = capsys.readouterr() assert exit_code == 75 assert exec_called is False assert json.loads(captured.out) == { "ok": False, "detail": "rate-limited", "reason": "all-accounts-capped", "resume_at": "2026-08-10T12:40:53Z", } def test_codex_router_explicit_override_refuses_broken_account_when_over_cap( home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: write_rules( home, { "filename": "routing_rules.json", "rules": { "projects": {}, "default": "rafa", "fallback_chain": [], "fallback_trigger": "broken_or_quota_exhausted", "quota_exhausted_threshold_pct": 100, "account_caps": {"rafa": {"7d": 10}}, }, }, ) tray = home / ".local/state/overdeck/systray/runtime" (tray / "accounts" / "rafa" / "CODEX_HOME").mkdir(parents=True) write_health( home, { "rafa": { "status": "broken", "primary_used_pct": 100, "secondary_used_pct": 91, "secondary_reset_at": 1786365653.0, }, }, "health_cache.json", ) router = command_router.CommandRouter(command_router.CodexAdapter()) from routing_resolver import AllAccountsCappedError resolved = router._resolve_account("rafa") assert isinstance(resolved, AllAccountsCappedError) exec_called = False def fake_execvpe(file: str, args: list[str], env: dict[str, str]) -> None: nonlocal exec_called exec_called = True monkeypatch.setattr(command_router.os, "execvpe", fake_execvpe) exit_code = router.main(["cdx", "--account", "rafa", "chat"]) captured = capsys.readouterr() assert exit_code == 75 assert exec_called is False assert json.loads(captured.out) == { "ok": False, "detail": "rate-limited", "reason": "all-accounts-capped", "resume_at": "2026-08-10T12:40:53Z", } def test_codex_router_explicit_override_resolves_broken_account_when_under_cap( home: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: write_rules( home, { "filename": "routing_rules.json", "rules": { "projects": {}, "default": "rafa", "fallback_chain": [], "fallback_trigger": "broken_or_quota_exhausted", "quota_exhausted_threshold_pct": 100, "account_caps": {"rafa": {"7d": 10}}, }, }, ) tray = home / ".local/state/overdeck/systray/runtime" (tray / "accounts" / "rafa" / "CODEX_HOME").mkdir(parents=True) write_health( home, { "rafa": { "status": "broken", "primary_used_pct": 100, "secondary_used_pct": 80, }, }, "health_cache.json", ) router = command_router.CommandRouter(command_router.CodexAdapter()) resolved = router._resolve_account("rafa") assert resolved == (tray, "rafa", tray / "accounts" / "rafa" / "CODEX_HOME") exec_call: dict[str, object] = {} monkeypatch.setattr( command_router.CodexAdapter, "sync_before_exec", lambda self, base_dir: None, ) monkeypatch.setattr(command_router.os, "execvpe", lambda *args, **kwargs: pytest.fail("execvpe should not run")) exit_code = router.main(["cdx", "--account", "rafa", "chat"]) assert exit_code == remote_dispatch.EXIT_UNCONTAINABLE def test_codex_router_explicit_override_resolves_quota_exhausted_without_caps( home: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: write_rules( home, { "filename": "routing_rules.json", "rules": { "projects": {}, "default": "rafa", "fallback_chain": [], "fallback_trigger": "broken_or_quota_exhausted", "quota_exhausted_threshold_pct": 100, }, }, ) tray = home / ".local/state/overdeck/systray/runtime" (tray / "accounts" / "rafa" / "CODEX_HOME").mkdir(parents=True) write_health( home, { "rafa": { "status": "ok", "primary_used_pct": 100, "secondary_used_pct": 100, }, }, "health_cache.json", ) router = command_router.CommandRouter(command_router.CodexAdapter()) resolved = router._resolve_account("rafa") assert resolved == (tray, "rafa", tray / "accounts" / "rafa" / "CODEX_HOME") exec_call: dict[str, object] = {} monkeypatch.setattr(command_router.os, "execvpe", lambda *args, **kwargs: pytest.fail("execvpe should not run")) monkeypatch.setattr( command_router.CodexAdapter, "sync_before_exec", lambda self, base_dir: None, ) exit_code = router.main(["cdx", "--account", "rafa", "chat"]) assert exit_code == remote_dispatch.EXIT_UNCONTAINABLE def test_claude_router_outside_container_returns_uncontainable_without_exec( home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: write_rules( home, { "filename": "claude_routing_rules.json", "rules": { "projects": {}, "default": "avi", "fallback_chain": [], "fallback_trigger": "broken_or_quota_exhausted", "quota_exhausted_threshold_pct": 100, }, }, ) claude_home = home / ".local/state/overdeck/systray/runtime" / "claude-accounts" / "avi" / "CLAUDE_HOME" claude_home.mkdir(parents=True) write_health( home, { "avi": {"status": "ok", "primary_used_pct": 10, "secondary_used_pct": 20}, }, "claude_health_cache.json", ) monkeypatch.setattr(command_router, "detect_project_name", lambda cwd=None: "unknown") monkeypatch.setattr(command_router.ClaudeAdapter, "verify_env_support", lambda self, home: True) router = command_router.CommandRouter( command_router.ClaudeAdapter(), detect_project_name=command_router.detect_project_name, ) monkeypatch.setattr( command_router.remote_dispatch, "in_container", lambda: False, ) exec_called = False def fake_execvpe(file: str, args: list[str], env: dict[str, str]) -> None: nonlocal exec_called exec_called = True monkeypatch.setattr(command_router.os, "execvpe", fake_execvpe) exit_code = router.main(["cld", "chat"]) captured = capsys.readouterr() assert exit_code == remote_dispatch.EXIT_UNCONTAINABLE assert exec_called is False assert captured.err.strip() == "cld: no containerized path for this invocation (adapter captures no log for these arguments)" def test_cdx_main_version_probe_execs_without_containment( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: router = command_router.CommandRouter( command_router.CodexAdapter(), detect_project_name=command_router.detect_project_name, ) account_home = tmp_path / "accounts" / "acct" / "CODEX_HOME" account_home.mkdir(parents=True) monkeypatch.setenv("OD_CONTAINMENT_LOG", str(tmp_path / "containment.jsonl")) monkeypatch.setattr( router, "_resolve_account", lambda *args, **kwargs: (tmp_path, "acct", account_home), ) monkeypatch.setattr(command_router.remote_dispatch, "in_container", lambda: False) monkeypatch.setattr( command_router.remote_dispatch, "open_session", lambda *args, **kwargs: pytest.fail("a --version probe must not open a remote session"), ) execed: list[list[str]] = [] monkeypatch.setattr( command_router.os, "execvpe", lambda file, args, env: execed.append(args), ) exit_code = router.main(["cdx", "--version"]) assert exit_code == 0 assert execed == [["codex", "--version"]] record = json.loads((tmp_path / "containment.jsonl").read_text(encoding="utf-8").splitlines()[-1]) assert record["mode"] == "info-only" assert record["containment"] == "none" def test_cdx_main_exec_with_version_flag_is_still_contained( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], ) -> None: router = command_router.CommandRouter( command_router.CodexAdapter(), detect_project_name=command_router.detect_project_name, ) account_home = tmp_path / "accounts" / "acct" / "CODEX_HOME" account_home.mkdir(parents=True) monkeypatch.setenv("OD_CONTAINMENT_LOG", str(tmp_path / "containment.jsonl")) monkeypatch.setattr( router, "_resolve_account", lambda *args, **kwargs: (tmp_path, "acct", account_home), ) monkeypatch.setattr(command_router.remote_dispatch, "in_container", lambda: False) monkeypatch.setattr( command_router.CodexAdapter, "capture_log_path", lambda self, forwarded_argv: None, ) monkeypatch.setattr( command_router.os, "execvpe", lambda file, args, env: pytest.fail("an agent invocation must not exec uncontained"), ) exit_code = router.main(["cdx", "exec", "--version"]) assert exit_code == remote_dispatch.EXIT_UNCONTAINABLE assert "no containerized path" in capsys.readouterr().err