from __future__ import annotations import json import stat from pathlib import Path from types import SimpleNamespace from typing import Any import httpx import pytest from gateway.platforms import dualcoach_admin as admin class _Models: def __init__(self, outcome: object) -> None: self.outcome = outcome self.calls: list[float] = [] def list(self, *, timeout: float) -> object: self.calls.append(timeout) if isinstance(self.outcome, BaseException): raise self.outcome return self.outcome class _Client: def __init__(self, outcome: object = SimpleNamespace(data=[])) -> None: self.models = _Models(outcome) def _config( *, provider: str = "openrouter", model: str = "coach-model", custom_providers: object | None = None, secret: str | None = None, ) -> dict[str, object]: coaching: dict[str, object] = { "draft_provider": provider, "draft_model": model, } if secret is not None: coaching["api_key"] = secret payload: dict[str, object] = {"physique_coach": coaching} if custom_providers is not None: payload["custom_providers"] = custom_providers return payload def test_provider_auth_uses_the_production_draft_model_default() -> None: assert admin._strict_config({"physique_coach": {}}) == admin.ProviderAuthConfig( provider="openai-codex", model="gpt-5.6-terra", custom_secret_reference=None, ) def _check( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, config: object, resolver: object, *, allow_billable_active_probe: bool = False, ) -> admin.ProviderAuthReceipt: monkeypatch.setattr(admin, "_load_production_config", lambda: config) monkeypatch.setattr(admin, "_resolve_provider_client", resolver) return admin.provider_auth_check( receipt_directory=tmp_path / "provider-auth", timestamp_utc="2026-08-14T12:00:00Z", allow_billable_active_probe=allow_billable_active_probe, ) def test_configured_supported_provider_uses_production_resolution_and_models_probe( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: client = _Client() calls: list[tuple[str, str]] = [] def resolver(provider: str, model: str) -> tuple[object, str]: calls.append((provider, model)) return client, model receipt = _check(monkeypatch, tmp_path, _config(), resolver) assert receipt.result is admin.ProviderAuthResult.READY assert receipt.exit is admin.ProviderAuthExit.READY assert calls == [("openrouter", "coach-model")] assert client.models.calls == [admin.PROBE_TIMEOUT_SECONDS] assert receipt.payload["probe"] == "models.list" assert receipt.payload["provider_adapter"] == admin.PROVIDER_ADAPTER assert receipt.payload["provider_adapter_version"] == admin.PROVIDER_ADAPTER_VERSION assert receipt.payload["success"] is True assert receipt.payload["command"] == "dualcoach_admin provider-auth check" assert receipt.payload["timestamp_utc"] == "2026-08-14T12:00:00Z" candidate_digest = receipt.payload["candidate_digest"] config_digest = receipt.payload["config_sha256"] assert isinstance(candidate_digest, str) and len(candidate_digest) == 64 assert isinstance(config_digest, str) and len(config_digest) == 64 assert receipt.receipt_path is not None assert receipt.index_path is not None assert stat.S_IMODE(receipt.receipt_path.stat().st_mode) == 0o600 assert stat.S_IMODE(receipt.index_path.stat().st_mode) == 0o600 assert json.loads(receipt.receipt_path.read_text(encoding="utf-8")) == receipt.payload @pytest.mark.parametrize( ("config", "expected_result", "expected_exit"), ( ({}, admin.ProviderAuthResult.CONFIG_MISSING, admin.ProviderAuthExit.CONFIG_MISSING), ( {"physique_coach": {"draft_provider": "openrouter", "draft_model": "coach-model"}}, admin.ProviderAuthResult.CREDENTIAL_MISSING, admin.ProviderAuthExit.CREDENTIAL_MISSING, ), ( _config(provider="not-a-provider"), admin.ProviderAuthResult.PROVIDER_UNRESOLVED, admin.ProviderAuthExit.PROVIDER_UNRESOLVED, ), ( _config(provider="bedrock"), admin.ProviderAuthResult.PROVIDER_UNSUPPORTED, admin.ProviderAuthExit.PROVIDER_UNSUPPORTED, ), ), ) def test_provider_auth_fails_closed_for_missing_and_unroutable_configuration( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, config: object, expected_result: admin.ProviderAuthResult, expected_exit: admin.ProviderAuthExit, ) -> None: calls: list[tuple[str, str]] = [] def resolver(provider: str, model: str) -> tuple[None, None]: calls.append((provider, model)) return None, None receipt = _check(monkeypatch, tmp_path, config, resolver) assert receipt.result is expected_result assert receipt.exit is expected_exit assert receipt.payload["success"] is False if expected_result in { admin.ProviderAuthResult.CONFIG_MISSING, admin.ProviderAuthResult.PROVIDER_UNRESOLVED, admin.ProviderAuthResult.PROVIDER_UNSUPPORTED, }: assert calls == [] else: assert calls == [("openrouter", "coach-model")] @pytest.mark.parametrize( ("secret_ref", "environment_present", "expected_result", "expected_exit"), ( ( "not/a/secret/reference", True, admin.ProviderAuthResult.SECRET_REFERENCE_MALFORMED, admin.ProviderAuthExit.SECRET_REFERENCE_MALFORMED, ), ( "DUALCOACH_TEST_MISSING_SECRET", False, admin.ProviderAuthResult.SECRET_SOURCE_UNAVAILABLE, admin.ProviderAuthExit.SECRET_SOURCE_UNAVAILABLE, ), ), ) def test_provider_auth_validates_real_custom_provider_secret_references_before_resolution( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, secret_ref: str, environment_present: bool, expected_result: admin.ProviderAuthResult, expected_exit: admin.ProviderAuthExit, ) -> None: if environment_present: monkeypatch.setenv("DUALCOACH_TEST_MISSING_SECRET", "test-secret") else: monkeypatch.delenv("DUALCOACH_TEST_MISSING_SECRET", raising=False) config = _config( provider="coach-private-route", custom_providers=[ { "name": "coach-private-route", "base_url": "https://provider.example/v1", "key_env": secret_ref, } ], ) def resolver(_provider: str, _model: str) -> tuple[object, str]: raise AssertionError("invalid secret configuration must not resolve a client") receipt = _check(monkeypatch, tmp_path, config, resolver) assert receipt.result is expected_result assert receipt.exit is expected_exit def test_provider_auth_rejects_provider_authentication_response_without_error_prose( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: class Rejected(RuntimeError): status_code = 401 secret = "token-value-that-must-never-appear" receipt = _check( monkeypatch, tmp_path, _config(secret=secret), lambda _provider, _model: (_Client(Rejected(secret)), "coach-model"), ) assert receipt.result is admin.ProviderAuthResult.AUTH_REJECTED assert receipt.exit is admin.ProviderAuthExit.AUTH_REJECTED serialized = json.dumps(receipt.payload, sort_keys=True) assert secret not in serialized assert "Rejected" not in serialized @pytest.mark.parametrize( ("outcome", "expected_result", "expected_exit"), ( ( TimeoutError("token-value-that-must-never-appear"), admin.ProviderAuthResult.PROBE_TIMEOUT, admin.ProviderAuthExit.PROBE_TIMEOUT, ), ( SimpleNamespace(data="not-a-list"), admin.ProviderAuthResult.PROBE_UNKNOWN, admin.ProviderAuthExit.PROBE_UNKNOWN, ), ), ) def test_provider_auth_timeout_and_unknown_probe_responses_fail_closed( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, outcome: object, expected_result: admin.ProviderAuthResult, expected_exit: admin.ProviderAuthExit, ) -> None: receipt = _check( monkeypatch, tmp_path, _config(), lambda _provider, _model: (_Client(outcome), "coach-model"), ) assert receipt.result is expected_result assert receipt.exit is expected_exit assert receipt.payload["success"] is False def test_config_hash_excludes_secrets_and_cli_output_is_redacted( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str], ) -> None: secret_one = "api-key-one-must-not-appear" secret_two = "api-key-two-must-not-appear" hashes: list[str] = [] for secret in (secret_one, secret_two): receipt = _check( monkeypatch, tmp_path / secret[-3:], _config(secret=secret), lambda _provider, _model: (_Client(), "coach-model"), ) hashes.append(str(receipt.payload["config_sha256"])) assert hashes[0] == hashes[1] monkeypatch.setattr(admin, "_load_production_config", lambda: _config(secret=secret_one)) monkeypatch.setattr( admin, "_resolve_provider_client", lambda _provider, _model: (_Client(RuntimeError(secret_two)), "coach-model"), ) exit_code = admin.main( [ "provider-auth", "check", "--json", "--receipt-dir", str(tmp_path / "cli-receipts"), ] ) output = capsys.readouterr().out assert exit_code == int(admin.ProviderAuthExit.PROBE_UNKNOWN) assert secret_one not in output assert secret_two not in output assert json.loads(output)["result"] == admin.ProviderAuthResult.PROBE_UNKNOWN.value class _CodexCompletions: def __init__(self, outcome: object) -> None: self.outcome = outcome self.calls: list[dict[str, object]] = [] def create(self, **kwargs: object) -> object: self.calls.append(kwargs) if isinstance(self.outcome, BaseException): raise self.outcome return self.outcome class _CodexClient: def __init__(self, outcome: object) -> None: self.chat = SimpleNamespace(completions=_CodexCompletions(outcome)) self.close_calls = 0 @property def calls(self) -> list[dict[str, object]]: return self.chat.completions.calls def close(self) -> None: self.close_calls += 1 def _codex_response(*, valid_contract: bool = True, status: str = "completed", usage: object | None = None) -> object: request_fields = ["input", "instructions", "model", "store", "stream", "timeout"] if not valid_contract: request_fields.append("metadata") return SimpleNamespace( model="gpt-5.1-codex-mini", usage=usage or SimpleNamespace(prompt_tokens=2, completion_tokens=1, total_tokens=3), provider_status=status, provider_terminal_received=True, provider_terminal_event_type="response.completed", provider_request_audit={ "endpoint_capability": "chatgpt_codex_responses_v1", "field_names": request_fields, "field_shapes": { "input": "array", "instructions": "string", "model": "string", "store": "boolean", "stream": "boolean", "timeout": "number", }, "strict_text_format": False, "requested_max_output_tokens": None, "max_output_tokens_sent": False, "store_is_false": True, "stream_is_true": True, }, ) def _codex_setup( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, outcome: object, *, local_status: object = None, ) -> _CodexClient: client = _CodexClient(outcome) monkeypatch.setattr(admin, "_is_codex_auxiliary_client", lambda value: value is client) monkeypatch.setattr(admin, "_configure_codex_one_shot", lambda value: value is client) monkeypatch.setattr( admin, "_load_codex_auth_status", lambda: {"logged_in": True} if local_status is None else local_status, ) snapshots = iter(("a" * 64, "a" * 64)) monkeypatch.setattr(admin, "_profile_snapshot_sha256", lambda: next(snapshots)) monkeypatch.setattr( admin, "_resolve_provider_client", lambda provider, model: ( client if (provider, model) == ("openai-codex", "coach-model") else None, "coach-model", ), ) return client def _codex_sdk_client( monkeypatch: pytest.MonkeyPatch, *, model: str, handler: Any, ) -> tuple[Any, list[httpx.Request]]: from agent.auxiliary_client import CodexAuxiliaryClient from openai import OpenAI requests: list[httpx.Request] = [] def capture(request: httpx.Request) -> httpx.Response: requests.append(request) return handler(request) raw_client = OpenAI( api_key="test-only", base_url="https://chatgpt.com/backend-api/codex", http_client=httpx.Client(transport=httpx.MockTransport(capture)), ) client = CodexAuxiliaryClient(raw_client, model) monkeypatch.setattr(admin, "_load_codex_auth_status", lambda: {"logged_in": True}) snapshots = iter(("a" * 64, "a" * 64)) monkeypatch.setattr(admin, "_profile_snapshot_sha256", lambda: next(snapshots)) monkeypatch.setattr( admin, "_resolve_provider_client", lambda provider, configured_model: (client, configured_model) if provider == "openai-codex" and configured_model == model else (None, None), ) return client, requests def test_codex_active_probe_rejects_known_retired_model_before_any_request( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: resolved: list[tuple[str, str]] = [] monkeypatch.setattr(admin, "_load_codex_auth_status", lambda: {"logged_in": True}) snapshots = iter(("a" * 64, "a" * 64)) monkeypatch.setattr(admin, "_profile_snapshot_sha256", lambda: next(snapshots)) monkeypatch.setattr( admin, "_resolve_provider_client", lambda provider, model: resolved.append((provider, model)) or (None, None), ) receipt = _check( monkeypatch, tmp_path, _config(provider="openai-codex", model="gpt-5.1-codex-mini"), admin._resolve_provider_client, allow_billable_active_probe=True, ) assert receipt.result is admin.ProviderAuthResult.CODEX_MODEL_UNSUPPORTED assert receipt.exit is admin.ProviderAuthExit.CODEX_MODEL_UNSUPPORTED assert resolved == [] assert receipt.payload["billable"] is False assert receipt.payload["model_preflight"] == "known_retired_codex_oauth_model" assert receipt.payload["request_attempts"] == 0 assert receipt.payload["sdk_max_retries"] is None def test_codex_active_probe_real_sdk_fake_transport_proves_one_shot_success( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: def handler(request: httpx.Request) -> httpx.Response: return httpx.Response( 200, request=request, headers={"content-type": "text/event-stream"}, content=( "event: response.created\n" 'data: {"type":"response.created","response":{"id":"resp_fake","status":"in_progress"}}\n\n' "event: response.output_item.done\n" 'data: {"type":"response.output_item.done","item":{"id":"msg_fake","type":"message","role":"assistant","status":"completed","content":[{"type":"output_text","text":"ok","annotations":[]}]}}\n\n' "event: response.completed\n" 'data: {"type":"response.completed","response":{"id":"resp_fake","object":"response","created_at":0,"status":"completed","model":"gpt-5.4","output":null,"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}\n\n' ), ) client, requests = _codex_sdk_client(monkeypatch, model="gpt-5.4", handler=handler) receipt = _check( monkeypatch, tmp_path, _config(provider="openai-codex", model="gpt-5.4"), admin._resolve_provider_client, allow_billable_active_probe=True, ) assert receipt.result is admin.ProviderAuthResult.READY assert len(requests) == 1 assert client._real_client.max_retries == 0 request_body = json.loads(requests[0].content) assert request_body["store"] is False assert request_body["stream"] is True assert set(request_body) == {"input", "instructions", "model", "store", "stream"} assert receipt.payload["request_attempts"] == 1 assert receipt.payload["sdk_max_retries"] == 0 assert receipt.payload["profile_snapshot_unchanged"] is True def test_codex_active_probe_real_sdk_fake_transport_classifies_retired_model_without_prose( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: secret = "provider-prose-that-must-not-escape" def handler(request: httpx.Request) -> httpx.Response: return httpx.Response( 400, request=request, json={ "error": { "message": secret, "code": "model_not_found", "type": "invalid_request_error", "param": "model", } }, ) client, requests = _codex_sdk_client(monkeypatch, model="gpt-5.4", handler=handler) receipt = _check( monkeypatch, tmp_path, _config(provider="openai-codex", model="gpt-5.4"), admin._resolve_provider_client, allow_billable_active_probe=True, ) assert receipt.result is admin.ProviderAuthResult.CODEX_MODEL_UNSUPPORTED assert receipt.exit is admin.ProviderAuthExit.CODEX_MODEL_UNSUPPORTED assert len(requests) == 1 assert client._real_client.max_retries == 0 assert receipt.payload["request_attempts"] == 1 assert receipt.payload["sdk_max_retries"] == 0 assert receipt.payload["request_contract"] == { "store": False, "stream": True, "tools": False, "metadata": False, "previous_response_id": False, "conversation_ids": False, "thread_ids": False, "sdk_max_retries": 0, "request_attempts": 1, } assert receipt.payload["provider_failure"] == { "http_status": 400, "provider_error_code": "model_not_found", "message_category": "bad_request", "failure_code": "provider_bad_request", "retryable": False, } assert secret not in json.dumps(receipt.payload, sort_keys=True) def test_codex_active_probe_requires_explicit_flag_and_closes_without_a_request( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: client = _codex_setup(monkeypatch, tmp_path, _codex_response()) receipt = _check( monkeypatch, tmp_path, _config(provider="openai-codex"), admin._resolve_provider_client, ) assert receipt.result is admin.ProviderAuthResult.PROBE_UNKNOWN assert receipt.exit is admin.ProviderAuthExit.PROBE_UNKNOWN assert client.calls == [] assert client.close_calls == 1 assert receipt.payload["billable"] is False assert receipt.payload["probe_kind"] == "none" def test_codex_active_probe_is_one_shot_nonpersistent_and_redacted( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: client = _codex_setup(monkeypatch, tmp_path, _codex_response()) receipt = _check( monkeypatch, tmp_path, _config(provider="openai-codex"), admin._resolve_provider_client, allow_billable_active_probe=True, ) assert receipt.result is admin.ProviderAuthResult.READY assert receipt.exit is admin.ProviderAuthExit.READY assert client.calls == [ { "messages": [{"role": "user", "content": "ok"}], "stream": True, "timeout": admin.PROBE_TIMEOUT_SECONDS, } ] assert client.close_calls == 1 assert receipt.payload["schema"] == "dualcoach-provider-auth-receipt-v2" assert receipt.payload["probe_kind"] == "codex_nonpersistent_generation" assert receipt.payload["store"] is False assert receipt.payload["billable"] is True assert receipt.payload["prompt_length"] == 2 assert receipt.payload["response_status"] == "completed" assert receipt.payload["sdk_max_retries"] == 0 assert receipt.payload["request_attempts"] == 1 assert receipt.payload["usage"] == {"input_tokens": 2, "output_tokens": 1, "total_tokens": 3} assert receipt.payload["profile_snapshot_unchanged"] is True contract = admin._string_object_mapping(receipt.payload["request_contract"]) assert contract is not None assert contract["sdk_max_retries"] == 0 assert contract["request_attempts"] == 1 assert receipt.payload["effects"] == { "delivery_actions": 0, "registry_mutations": 0, "service_actions": 0, "telegram_actions": 0, } assert '"ok"' not in json.dumps(receipt.payload, sort_keys=True) @pytest.mark.parametrize( ("outcome", "expected_result", "expected_exit"), ( ( type("Rejected", (RuntimeError,), {"status_code": 401})("secret"), admin.ProviderAuthResult.AUTH_REJECTED, admin.ProviderAuthExit.AUTH_REJECTED, ), ( type("Limited", (RuntimeError,), {"status_code": 429})("secret"), admin.ProviderAuthResult.PROBE_RATE_LIMITED, admin.ProviderAuthExit.PROBE_RATE_LIMITED, ), ( TimeoutError("secret"), admin.ProviderAuthResult.PROBE_TIMEOUT, admin.ProviderAuthExit.PROBE_TIMEOUT, ), ( InterruptedError("secret"), admin.ProviderAuthResult.PROBE_CANCELLED, admin.ProviderAuthExit.PROBE_CANCELLED, ), ( RuntimeError("secret"), admin.ProviderAuthResult.ACTIVE_PROBE_UNKNOWN, admin.ProviderAuthExit.ACTIVE_PROBE_UNKNOWN, ), ), ) def test_codex_active_probe_maps_failures_and_never_retries( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, outcome: object, expected_result: admin.ProviderAuthResult, expected_exit: admin.ProviderAuthExit, ) -> None: client = _codex_setup(monkeypatch, tmp_path, outcome) receipt = _check( monkeypatch, tmp_path, _config(provider="openai-codex"), admin._resolve_provider_client, allow_billable_active_probe=True, ) assert receipt.result is expected_result assert receipt.exit is expected_exit assert len(client.calls) == 1 assert client.close_calls == 1 assert "secret" not in json.dumps(receipt.payload, sort_keys=True) @pytest.mark.parametrize( "response", ( _codex_response(status="incomplete"), _codex_response(valid_contract=False), SimpleNamespace( model="", usage=SimpleNamespace(prompt_tokens=2, completion_tokens=1, total_tokens=3), provider_status="completed", provider_terminal_received=True, provider_terminal_event_type="response.completed", provider_request_audit={"field_names": []}, ), ), ) def test_codex_active_probe_rejects_incomplete_or_malformed_response( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, response: object, ) -> None: client = _codex_setup(monkeypatch, tmp_path, response) receipt = _check( monkeypatch, tmp_path, _config(provider="openai-codex"), admin._resolve_provider_client, allow_billable_active_probe=True, ) assert receipt.result is admin.ProviderAuthResult.ACTIVE_PROBE_MALFORMED assert receipt.exit is admin.ProviderAuthExit.ACTIVE_PROBE_MALFORMED assert len(client.calls) == 1 assert client.close_calls == 1 @pytest.mark.parametrize( ("local_status", "expected_result", "expected_exit"), ( ({"logged_in": False}, admin.ProviderAuthResult.CREDENTIAL_MISSING, admin.ProviderAuthExit.CREDENTIAL_MISSING), ({"logged_in": True, "rate_limited": True}, admin.ProviderAuthResult.PROBE_RATE_LIMITED, admin.ProviderAuthExit.PROBE_RATE_LIMITED), ({}, admin.ProviderAuthResult.ACTIVE_PROBE_UNKNOWN, admin.ProviderAuthExit.ACTIVE_PROBE_UNKNOWN), ), ) def test_codex_active_probe_requires_local_auth_before_resolution( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, local_status: object, expected_result: admin.ProviderAuthResult, expected_exit: admin.ProviderAuthExit, ) -> None: client = _codex_setup(monkeypatch, tmp_path, _codex_response(), local_status=local_status) receipt = _check( monkeypatch, tmp_path, _config(provider="openai-codex"), admin._resolve_provider_client, allow_billable_active_probe=True, ) assert receipt.result is expected_result assert receipt.exit is expected_exit assert client.calls == [] assert client.close_calls == 0 def test_codex_active_probe_rebinds_the_production_adapter_to_zero_sdk_retries() -> None: from agent.auxiliary_client import CodexAuxiliaryClient from openai import OpenAI client = CodexAuxiliaryClient(OpenAI(api_key="test-only"), "gpt-5.1-codex-mini") try: assert admin._configure_codex_one_shot(client) is True assert client._real_client.max_retries == 0 assert client.chat.completions._client is client._real_client finally: client.close() def test_codex_active_probe_refuses_to_run_when_sdk_one_shot_cannot_be_attested( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: client = _codex_setup(monkeypatch, tmp_path, _codex_response()) monkeypatch.setattr(admin, "_configure_codex_one_shot", lambda _client: False) receipt = _check( monkeypatch, tmp_path, _config(provider="openai-codex"), admin._resolve_provider_client, allow_billable_active_probe=True, ) assert receipt.result is admin.ProviderAuthResult.ACTIVE_PROBE_UNKNOWN assert receipt.exit is admin.ProviderAuthExit.ACTIVE_PROBE_UNKNOWN assert client.calls == [] assert client.close_calls == 1 def test_cli_flag_is_required_to_enable_the_codex_active_probe( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, capsys: pytest.CaptureFixture[str], ) -> None: client = _codex_setup(monkeypatch, tmp_path, _codex_response()) monkeypatch.setattr(admin, "_load_production_config", lambda: _config(provider="openai-codex")) exit_code = admin.main( [ "provider-auth", "check", "--json", "--allow-billable-active-probe", "--receipt-dir", str(tmp_path / "cli-receipts"), ] ) assert exit_code == 0 assert len(client.calls) == 1 assert json.loads(capsys.readouterr().out)["probe_kind"] == "codex_nonpersistent_generation" def test_codex_active_probe_fails_closed_on_profile_drift_after_one_request( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: client = _codex_setup(monkeypatch, tmp_path, _codex_response()) snapshots = iter(("a" * 64, "b" * 64)) monkeypatch.setattr(admin, "_profile_snapshot_sha256", lambda: next(snapshots)) receipt = _check( monkeypatch, tmp_path, _config(provider="openai-codex"), admin._resolve_provider_client, allow_billable_active_probe=True, ) assert receipt.result is admin.ProviderAuthResult.PROFILE_MUTATED assert receipt.exit is admin.ProviderAuthExit.PROFILE_MUTATED assert len(client.calls) == 1 assert client.close_calls == 1 assert receipt.payload["profile_snapshot_unchanged"] is False def test_codex_active_probe_fails_closed_when_client_close_fails( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, ) -> None: client = _codex_setup(monkeypatch, tmp_path, _codex_response()) def close() -> None: client.close_calls += 1 raise RuntimeError("close failure with secret") monkeypatch.setattr(client, "close", close) receipt = _check( monkeypatch, tmp_path, _config(provider="openai-codex"), admin._resolve_provider_client, allow_billable_active_probe=True, ) assert receipt.result is admin.ProviderAuthResult.CLIENT_CLOSE_FAILED assert receipt.exit is admin.ProviderAuthExit.CLIENT_CLOSE_FAILED assert len(client.calls) == 1 assert client.close_calls == 1 assert "secret" not in json.dumps(receipt.payload, sort_keys=True)