"""Synthetic-only native endpoint tests; never import or start agent_server.""" import ast import asyncio import errno import json import os from pathlib import Path import tempfile import threading import tomllib from types import SimpleNamespace from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import unittest from unittest.mock import AsyncMock, Mock, patch from fastapi import FastAPI, HTTPException from fastapi.testclient import TestClient import codex_auth import codex_provider as provider from codex_app_server import CodexAppServerClient, CodexAppServerRequestError, CodexAppServerProtocolError from tests.test_codex_app_server import FakeProcessFactory, NO_RESPONSE import test_codex_auth_isolated as auth_fixture from test_codex_auth_isolated import NATIVE KEY = "provider-synthetic-secret-not-valid" SELECTION = {"base_url": "https://provider.example.invalid/v1", "model": "test/model", "api_key": KEY} class StoreTests(unittest.TestCase): def setUp(self): temporary = tempfile.TemporaryDirectory(prefix="provider-store-") self.addCleanup(temporary.cleanup) self.store = provider.ProviderStore(Path(temporary.name) / "private") def test_private_atomic_bound_credential_roundtrip_reset(self): self.assertFalse(self.store.status()["configured"]) self.store.save(SELECTION) self.assertEqual(self.store.selection(include_key=True), SELECTION) self.assertEqual(self.store.status()["credential_id"], self.store.revision()) self.assertEqual(self.store.catalog(available=True)["credential_id"], self.store.revision()) self.assertNotIn(KEY, json.dumps(self.store.status())) self.assertNotIn(KEY, (self.store.root / "settings.json").read_text()) for path in self.store.root.iterdir(): self.assertEqual(path.stat().st_mode & 0o777, 0o600) old = self.store.revision() self.store.save({**SELECTION, "api_key": "replacement-synthetic"}) self.assertTrue((self.store.root / ("credential-" + old + ".json")).exists()) self.assertEqual(self.store.selection(revision=old, include_key=True)["api_key"], KEY) self.store.reset() self.assertFalse(self.store.status()["configured"]) def test_thread_binding_prevents_cross_endpoint_and_default_history(self): self.store.save(SELECTION) with self.assertRaises(HTTPException): self.store.require_thread("old-default-thread", SELECTION) self.store.record_thread("custom-thread/../../not-a-path", SELECTION) self.store.require_thread("custom-thread/../../not-a-path", SELECTION) self.store.require_thread("custom-thread/../../not-a-path", {**SELECTION, "model": "different"}) with self.assertRaises(HTTPException): self.store.require_thread("custom-thread/../../not-a-path", {**SELECTION, "base_url": "https://other.example.invalid/v1"}) self.store.reset() with self.assertRaises(HTTPException): self.store.require_thread("custom-thread/../../not-a-path", None) self.store.require_thread("old-default-thread", None) def test_model_capability_evidence_survives_refresh_without_crossing_credentials(self): self.store.save(SELECTION) listed = {"models": [{"value": "test/model", "label": "test/model"}, {"value": "other/model", "label": "other/model"}], "model_capabilities": {"test/model": {"kind": "chat", "reasoning_efforts": ["low", "high"]}}} self.store.cache_catalog(SELECTION, listed) self.store.cache_model_capability(SELECTION, {"compatibility": "verified", "api_key": KEY}) self.store.cache_catalog(SELECTION, listed) catalog = self.store.cached_catalog(SELECTION) self.assertEqual(catalog["model_capabilities"]["test/model"], {"kind": "chat", "compatibility": "verified", "reasoning_efforts": ["low", "high"], "reasoning_supported": True, "reasoning_summary_supported": None}) self.assertEqual(catalog["model_capabilities"]["other/model"]["compatibility"], "unverified") self.assertEqual(catalog["model_efforts"]["other/model"], []) self.assertEqual((catalog["efforts"], catalog["default_effort"]), ([], "")) self.assertNotIn(KEY, json.dumps(self.store._model_capabilities)) revision = self.store.revision() self.store.save({**SELECTION, "api_key": "new-synthetic"}) self.assertEqual(self.store.cached_catalog()["model_capabilities"]["test/model"]["compatibility"], "unverified") self.assertEqual(self.store.catalog(available=True, session={"codex_provider": "custom", "codex_provider_revision": revision})["model_capabilities"]["test/model"]["compatibility"], "verified") other_endpoint = {**SELECTION, "base_url": "https://other.example.invalid/v1"} self.store.cache_catalog(other_endpoint, listed) self.assertEqual(self.store.cached_catalog(other_endpoint)["model_capabilities"]["test/model"]["compatibility"], "unverified") def test_summary_evidence_survives_discovery_without_crossing_models_or_credentials(self): self.store.save(SELECTION) selected = self.store.registration(include_key=True) self.store.cache_model_capability(selected, {"reasoning_summary_supported": True}) self.store.cache_catalog(SELECTION, {"models": [{"value": "test/model"}, {"value": "other/model"}]}) catalog = self.store.cached_catalog(selected) self.assertEqual(provider.runtime_summary(SELECTION, catalog), "auto") self.assertEqual(provider.runtime_summary({**SELECTION, "model": "other/model"}, catalog), "none") # An inconclusive basic-check result supplies no summary evidence. self.store.cache_model_capability(selected, {"compatibility": "verified"}) self.assertEqual(provider.runtime_summary(selected, self.store.cached_catalog(selected)), "auto") self.store.save({**SELECTION, "api_key": "replacement-synthetic"}) self.assertEqual(provider.runtime_summary(SELECTION, self.store.cached_catalog()), "none") def test_discovered_efforts_reload_for_the_exact_saved_revision(self): self.store.save(SELECTION) selected = self.store.registration(include_key=True) self.store.cache_catalog(selected, {"models": [{"value": selected["model"]}], "default_model": selected["model"], "model_capabilities": {selected["model"]: { "reasoning_efforts": ["low", "high"], "compatibility": "verified", "reasoning_summary_supported": True}}}) record = self.store.root / ("model-catalog-" + selected["credential_id"] + ".json") self.assertEqual(record.stat().st_mode & 0o777, 0o600) self.assertNotIn(KEY, record.read_text()) self.store.save(SELECTION) replacement = self.store.registration(include_key=True) reloaded = provider.ProviderStore(self.store.root) current = reloaded.cached_catalog(replacement) self.assertEqual(provider.runtime_effort(replacement, current, "high"), "") retained = reloaded.cached_catalog(selected) self.assertEqual(provider.runtime_effort(selected, retained, "high"), "high") self.assertEqual(retained["models"], [{"value": selected["model"], "label": selected["model"]}]) self.assertEqual(retained["model_capabilities"][selected["model"]]["compatibility"], "unverified") self.assertEqual(provider.runtime_summary(selected, retained), "none") record.write_text("invalid optional catalog") unavailable = provider.ProviderStore(self.store.root).cached_catalog(selected) self.assertEqual(provider.runtime_effort(selected, unavailable, "high"), "") def test_summary_evidence_reloads_privately_without_persisting_basic_compatibility(self): self.store.save(SELECTION) selected = self.store.registration(include_key=True) self.store.cache_model_capability(selected, {"compatibility": "verified", "reasoning_summary_supported": True}) record = self.store.root / ("summary-capabilities-" + selected["credential_id"] + ".json") self.assertEqual(record.stat().st_mode & 0o777, 0o600) self.assertNotIn(KEY, record.read_text()) reloaded = provider.ProviderStore(self.store.root) catalog = reloaded.cached_catalog() self.assertEqual(provider.runtime_summary(selected, catalog), "auto") self.assertEqual(catalog["model_capabilities"][selected["model"]]["compatibility"], "unverified") self.assertEqual(provider.runtime_summary({**selected, "model": "unverified/other"}, catalog), "none") def test_summary_evidence_is_revision_scoped_and_fresh_rejection_survives_reset(self): self.store.save(SELECTION) original = self.store.registration(include_key=True) self.store.cache_model_capability(original, {"reasoning_summary_supported": True}) # Even a new revision of identical credentials starts without old proof. self.store.save(SELECTION) replacement = self.store.registration(include_key=True) self.assertNotEqual(original["credential_id"], replacement["credential_id"]) self.assertEqual(provider.runtime_summary(replacement, self.store.cached_catalog()), "none") self.assertEqual(provider.runtime_summary(original, self.store.cached_catalog(original)), "auto") self.store.cache_model_capability(original, {"reasoning_summary_supported": False}) self.store.cache_model_capability(original, {"compatibility": "verified", "reasoning_summary_supported": None}) self.store.reset() reloaded = provider.ProviderStore(self.store.root) self.assertFalse(reloaded.status()["configured"]) retained = reloaded.selection(include_key=True, revision=original["credential_id"]) capability = reloaded.cached_catalog(retained)["model_capabilities"][retained["model"]] self.assertIs(capability["reasoning_summary_supported"], False) self.assertEqual(provider.runtime_summary(retained, reloaded.cached_catalog(retained)), "none") def test_unsaved_probe_cannot_persist_or_override_saved_summary_evidence(self): self.store.save(SELECTION) selected = self.store.registration(include_key=True) self.store.cache_model_capability(selected, {"reasoning_summary_supported": True}) before = {p.name: p.read_bytes() for p in self.store.root.iterdir()} self.store.cache_model_capability(SELECTION, {"reasoning_summary_supported": False}) self.assertEqual(before, {p.name: p.read_bytes() for p in self.store.root.iterdir()}) self.assertEqual(provider.runtime_summary(selected, self.store.cached_catalog(selected)), "auto") with self.assertRaises(HTTPException): self.store.cache_model_capability({**selected, "api_key": "different-synthetic"}, {"reasoning_summary_supported": False}) self.assertEqual(provider.runtime_summary(selected, self.store.cached_catalog(selected)), "auto") def test_optional_summary_evidence_fails_closed_without_blocking_catalog_and_denials_win(self): self.store.save(SELECTION) selected = self.store.registration(include_key=True) self.store.cache_model_capability(selected, {"reasoning_summary_supported": True}) self.store.cache_catalog(selected, {"models": [{"value": selected["model"]}], "model_capabilities": {selected["model"]: {"reasoning_summary_supported": False}}}) self.assertEqual(provider.runtime_summary(selected, self.store.cached_catalog(selected)), "none") # A later successful explicit check supersedes older negative metadata. self.store.cache_model_capability(selected, {"reasoning_summary_supported": True}) self.assertEqual(provider.runtime_summary(selected, self.store.cached_catalog(selected)), "auto") self.store.cache_catalog(selected, {"models": [{"value": selected["model"]}], "model_capabilities": {selected["model"]: {"reasoning_summary_supported": False}}}) self.assertEqual(provider.runtime_summary(selected, self.store.cached_catalog(selected)), "none") self.assertEqual(provider.runtime_summary(selected, provider.ProviderStore(self.store.root).cached_catalog()), "none") self.store.cache_model_capability(selected, {"reasoning_summary_supported": False}) self.store.cache_catalog(selected, {"models": [{"value": selected["model"]}], "model_capabilities": {selected["model"]: {"reasoning_summary_supported": True}}}) self.assertEqual(provider.runtime_summary(selected, self.store.cached_catalog(selected)), "none") record = self.store.root / ("summary-capabilities-" + selected["credential_id"] + ".json") record.write_text("invalid optional metadata " + KEY) reloaded = provider.ProviderStore(self.store.root) self.assertTrue(reloaded.catalog(available=True)["configured"]) self.assertEqual(provider.runtime_summary(selected, reloaded.cached_catalog()), "none") # Unreadable/unsafe optional metadata is equally non-blocking. record.chmod(0o644) self.assertTrue(provider.ProviderStore(self.store.root).catalog(available=True)["configured"]) def test_summary_observation_capacity_retains_newest_without_rejecting_check(self): self.store.save(SELECTION) selected = self.store.registration(include_key=True) identifier = selected["credential_id"] previous = {f"model/{index}": True for index in range(512)} self.store._summary_capabilities[identifier] = previous self.store.cache_model_capability(selected, {"reasoning_summary_supported": True}) reloaded = provider.ProviderStore(self.store.root) catalog = reloaded.cached_catalog() self.assertEqual(provider.runtime_summary(selected, catalog), "auto") self.assertEqual(len(catalog["model_capabilities"]), 512) self.assertNotIn("model/0", catalog["model_capabilities"]) def test_legacy_binding_migrates_without_key_reentry_and_survives_reset(self): self.store.save(SELECTION) revision = self.store.revision() self.store._atomic("credential-" + revision + ".json", {"api_key": KEY, "binding": provider.legacy_binding(SELECTION)}) self.store._atomic(self.store._binding_name("legacy-thread"), {"binding": provider.legacy_binding(SELECTION)}) selected = self.store.for_session({"codex_provider": "custom", "codex_provider_binding": provider.legacy_binding(SELECTION), "model": "another-model"}) self.store.require_thread("legacy-thread", selected) self.store.save({"base_url": "https://next.example.invalid/v1", "api_key": "new-synthetic"}) self.store.reset() retained = self.store.for_thread("legacy-thread", include_key=True) self.assertEqual((retained["credential_id"], retained["api_key"]), (revision, KEY)) self.store.require_thread("legacy-thread", {**retained, "model": "third-model"}) def test_failure_before_metadata_commit_retains_old_key(self): self.store.save(SELECTION) atomic = self.store._atomic def failing(name, value): if name == "settings.json": raise OSError(errno.ENOSPC, "synthetic") return atomic(name, value) with patch.object(self.store, "_atomic", side_effect=failing): with self.assertRaises(OSError): self.store.save({**SELECTION, "api_key": "new-synthetic"}) self.assertEqual(self.store.selection(include_key=True), SELECTION) def test_directory_fsync_after_commit_never_dangles_selected_credential(self): self.store.save({**SELECTION, "api_key": "old-synthetic"}) previous = self.store.revision() fsync = os.fsync import stat def failing(fd): if stat.S_ISDIR(os.fstat(fd).st_mode): raise OSError(errno.EIO, "synthetic durability failure") fsync(fd) with patch.object(provider.os, "fsync", side_effect=failing): self.store.save(SELECTION) self.assertEqual(self.store.selection(include_key=True), SELECTION) self.assertTrue((self.store.root / ("credential-" + previous + ".json")).exists()) def test_binding_tamper_and_symlink_fail_closed(self): self.store.save(SELECTION) metadata_path = self.store.root / "settings.json" metadata = json.loads(metadata_path.read_text()) metadata["base_url"] = "https://different.example.invalid" self.store._atomic("settings.json", metadata) with self.assertRaises(HTTPException): self.store.selection(include_key=True) metadata_path.unlink() metadata_path.symlink_to(self.store.root / "missing") with self.assertRaises(HTTPException): self.store.save(SELECTION) def test_validation_and_native_args_never_include_secret(self): for base in ["http://example.invalid", "https://user:pass@example.invalid", "https://example.invalid/?token=x", "https://example.invalid/#secret", "https://example.invalid/v1/responses", "https://example.invalid/v1/chat/completions"]: with self.subTest(base=base), self.assertRaises(HTTPException): provider.validate_selection({**SELECTION, "base_url": base}) selected = provider.validate_selection({**SELECTION, "base_url": " http://127.0.0.1:7890/v1/ "}) self.assertEqual(selected["base_url"], "http://127.0.0.1:7890/v1") args = provider.native_args(SELECTION) self.assertNotIn(KEY, str(args)) self.assertIn("model_providers.agentsdock_custom.name=\"AgentsDock custom endpoint\"", args) self.assertFalse(any('."' in value.split("=", 1)[0] for value in args[1::2])) config = tomllib.loads("\n".join(args[1::2])) native = config["model_providers"][provider.PROVIDER_ID] self.assertFalse(native["requires_openai_auth"]) self.assertEqual(native["wire_api"], "responses") self.assertNotIn("request_max_retries", native) self.assertNotIn("stream_max_retries", native) self.assertFalse(any("max_retries=" in value for value in provider.registration_args(SELECTION))) env = provider.native_environment({"OPENAI_API_KEY": "old", "CODEX_API_KEY": "old", "https_proxy": "old", "PATH": "safe"}, SELECTION) self.assertEqual(env, {"PATH": "safe", provider.ENV_KEY: KEY, "RUST_LOG": "off"}) client = CodexAppServerClient("unused", cwd="/", env_factory=dict, sensitive_values=(KEY,)) self.assertTrue(client._authentication_submitted) self.assertNotIn(KEY, str(client._redact_sensitive({"error": {"message": KEY, "data": [KEY]}, KEY: KEY}))) class RouterTests(unittest.IsolatedAsyncioTestCase): def setUp(self): fixture = auth_fixture.CodexAuthTests() fixture.setUp() self.addCleanup(fixture.doCleanups) self.ns = fixture.ns temporary = tempfile.TemporaryDirectory(prefix="provider-router-") self.addCleanup(temporary.cleanup) self.store = provider.ProviderStore(Path(temporary.name) / "private") self.manager = fixture.manager self.manager.close = AsyncMock() self.ns.update({"codex_provider": provider, "CODEX_PROVIDER_STORE": self.store, "CODEX_PROVIDER_SETTINGS_LOCK": asyncio.Lock(), "STORE": SimpleNamespace(_lock=asyncio.Lock(), sessions={}, save=AsyncMock()), "session_codex_thread_id": lambda session: session.get("codex_thread_id", ""), "CODEX_APP_SERVER_MANAGER": self.manager, "CODEX_APP_SERVER_MANAGER_EPOCH": 1, "CODEX_APP_SERVER_MANAGER_LOCK": asyncio.Lock(), "CODEX_APP_SERVER_THREAD_LRU_LOCK": asyncio.Lock(), "CODEX_APP_SERVER_EVICTING_THREADS": {}, "CODEX_APP_SERVER_THREAD_LRU": {}, "CODEX_APP_SERVER_PINNED_THREADS": set(), "CODEX_APP_SERVER_THREAD_PIN_COUNTS": {}, "CODEX_APP_SERVER_INVALIDATED_THREADS": set(), "CODEX_APPROVAL_ITEM_CACHE": {}, "CODEX_PERMISSION_PROFILES_CACHE": {}, "RUNTIME_DIAGNOSTICS_LOCK": threading.RLock(), "RUNTIME_DIAGNOSTICS": {"codex": {"ready": False}, "claude": {"ready": True}}}) source = Path(__file__).with_name("agent_server.py") nodes = [node for node in ast.parse(source.read_text()).body if isinstance(node, ast.AsyncFunctionDef) and node.name in {"mutate_codex_provider", "replace_codex_provider_settings", "join_task_despite_caller_cancellation"}] from contextlib import suppress self.ns["suppress"] = suppress exec(compile(ast.Module(body=nodes, type_ignores=[]), str(source), "exec"), self.ns) self.probe = AsyncMock(return_value=provider.test_result("ready")) self.discover = Mock(return_value={**provider.test_result("ready"), "models": [{"value": "first/model", "label": "first/model"}], "default_model": "first/model"}) app = FastAPI() app.middleware("http")(self.ns["require_agent_token"]) app.include_router(provider.create_router(authorize=self.ns["require_native_admin_control"], store=self.store, mutate=self.ns["mutate_codex_provider"], probe=self.probe, available=lambda: True, discover=self.discover, session_lookup=lambda session_id: self.ns["STORE"].sessions.get(session_id))) app.include_router(codex_auth.create_router(authorize=self.ns["require_native_admin_control"], operation=self.ns["codex_auth_operation"], available=lambda: True, )) self.client = TestClient(app) self.addCleanup(self.client.close) def test_save_reset_preserve_manager_and_normal_readiness(self): response = self.client.put("/api/admin/codex/provider", headers=NATIVE, json=SELECTION) self.assertEqual(response.status_code, 200, response.text) self.assertTrue(response.json()["configured"]) self.assertNotIn(KEY, response.text) self.manager.close.assert_not_awaited() self.ns["close_codex_app_server_manager"].assert_not_awaited() self.assertEqual(self.ns["RUNTIME_DIAGNOSTICS"], {"codex": {"ready": False}, "claude": {"ready": True}}) self.assertFalse(self.ns["CODEX_GOALS_RECONFIGURING"]) self.ns["CODEX_APP_SERVER_MANAGER"] = self.manager response = self.client.delete("/api/admin/codex/provider", headers=NATIVE) self.assertEqual(response.status_code, 200, response.text) self.assertFalse(response.json()["configured"]) def test_optional_summary_write_failure_preserves_success_and_in_memory_evidence(self): self.store.save(SELECTION) selected = self.store.registration(include_key=True) self.probe.return_value = {**provider.test_result("ready"), "compatibility": "verified", "reasoning_summary_supported": True, "summary_check": "supported"} with patch.object(self.store, "_atomic", side_effect=OSError(errno.ENOSPC, KEY)): response = self.client.post("/api/admin/codex/provider/test", headers=NATIVE, json={"model": selected["model"], "credential_id": selected["credential_id"]}) self.assertEqual(response.status_code, 200) self.assertTrue(response.json()["ok"]) self.assertEqual(response.json()["summary_check"], "supported") self.assertNotIn(KEY, response.text) self.assertEqual(provider.runtime_summary(selected, self.store.cached_catalog(selected)), "auto") self.assertEqual(provider.runtime_summary(selected, provider.ProviderStore(self.store.root).cached_catalog()), "none") def test_custom_preserves_normal_login_and_busy_work_does_not_block_reset(self): self.store.save(SELECTION) response = self.client.post("/api/admin/codex/auth/api-key", headers=NATIVE, json={"api_key": KEY}) self.assertEqual(response.status_code, 409, response.text) self.manager.request.assert_not_awaited() self.assertEqual(self.store.selection(include_key=True), SELECTION) self.manager.client._turns_by_thread = {"native": SimpleNamespace(_completed=False)} response = self.client.delete("/api/admin/codex/provider", headers=NATIVE) self.assertEqual(response.status_code, 200, response.text) self.manager.close.assert_not_awaited() self.assertFalse(self.store.status()["configured"]) def test_save_pins_legacy_live_session_before_selecting_new_credentials(self): self.store.save(SELECTION) revision = self.store.revision() session = {"codex_provider": "custom", "codex_provider_binding": provider.legacy_binding(SELECTION), "codex_thread_id": "old-thread"} self.ns["STORE"].sessions["chat"] = session self.manager.client._turns_by_thread = {"old-thread": SimpleNamespace(_completed=False)} response = self.client.put("/api/admin/codex/provider", headers=NATIVE, json={"base_url": SELECTION["base_url"], "api_key": "new-synthetic"}) self.assertEqual(response.status_code, 200, response.text) self.assertEqual(session["codex_provider_revision"], revision) self.assertEqual(self.store.for_session(session, include_key=True)["api_key"], KEY) self.assertEqual(self.store.for_thread("old-thread")["credential_id"], revision) self.ns["STORE"].save.assert_awaited_once_with(durable=True) self.manager.close.assert_not_awaited() def test_endpoint_only_save_and_explicit_discovery_cache_per_credential(self): selected = {"base_url": SELECTION["base_url"], "api_key": KEY} tested = self.client.post("/api/admin/codex/provider/test", headers=NATIVE, json=selected) self.assertEqual(tested.status_code, 200, tested.text) self.probe.assert_not_awaited() self.assertFalse(self.store.root.exists()) response = self.client.put("/api/admin/codex/provider", headers=NATIVE, json=selected) self.assertEqual(response.status_code, 200, response.text) self.assertIsNone(response.json()["model"]) self.assertEqual(self.store.catalog(available=True)["default_model"], "first/model") self.assertEqual(self.discover.call_count, 1) response = self.client.get("/api/admin/codex/provider/models", headers=NATIVE) self.assertEqual(response.status_code, 200, response.text) self.assertEqual(self.discover.call_count, 2) self.store.save({**selected, "api_key": "different-synthetic"}) self.assertEqual(self.store.catalog(available=True)["models"], []) def test_model_refresh_uses_retained_chat_credentials_after_reset(self): self.store.save(SELECTION) session = {"codex_provider": "custom", "codex_provider_revision": self.store.revision(), "model": "chat/model"} self.ns["STORE"].sessions["chat"] = session self.store.reset() received = [] def discover(selected): received.append(dict(selected)) return {"ok": True, "status": "ready", "models": [{"value": "retained/model", "label": "retained/model"}]} self.discover.side_effect = discover response = self.client.get("/api/admin/codex/provider/models?session_id=chat", headers=NATIVE) self.assertEqual(response.status_code, 200, response.text) self.assertEqual(received[0]["api_key"], KEY) self.assertNotIn(KEY, response.text) self.assertEqual(self.store.catalog(available=True, session=session)["models"][0]["value"], "retained/model") self.assertTrue(self.store.catalog(available=True, session=session)["available"]) self.assertFalse(self.store.catalog(available=True)["available"]) with patch.object(self.store, "_read", side_effect=AssertionError("projection must not read credentials")): summary = self.store.catalog(available=True, session=session, summary=True) self.assertTrue(summary["available"]) self.assertNotIn("models", summary) self.assertNotIn(KEY, json.dumps(summary)) def test_reset_with_missing_credential_never_initializes_a_provider(self): self.store.save(SELECTION) identifier = self.store.revision() (self.store.root / ("credential-" + identifier + ".json")).unlink() self.ns["CODEX_APP_SERVER_MANAGER"] = None response = self.client.delete("/api/admin/codex/provider", headers=NATIVE) self.assertEqual(response.status_code, 200, response.text) self.assertFalse(response.json()["configured"]) self.ns["codex_app_server_manager"].assert_not_awaited() def test_auth_origin_transport_and_validation_fail_before_probe(self): for headers, expected in [({}, 401), ({**NATIVE, "Origin": "https://example.invalid"}, 403), ({**NATIVE, "Content-Type": "text/plain"}, 415)]: response = self.client.post("/api/admin/codex/provider/test", headers=headers, content=json.dumps(SELECTION)) self.assertEqual(response.status_code, expected, response.text) self.assertNotIn(KEY, response.text) response = self.client.post("/api/admin/codex/provider/test", headers=NATIVE, json={**SELECTION, "api_key": [KEY]}) self.assertEqual(response.status_code, 400, response.text) self.assertNotIn(KEY, response.text) self.probe.assert_not_awaited() def test_test_isolated_no_store_write_and_safe_errors(self): response = self.client.post("/api/admin/codex/provider/test", headers=NATIVE, json=SELECTION) self.assertEqual(response.status_code, 200, response.text) self.assertEqual(response.json()["status"], "ready") self.assertFalse(self.store.root.exists()) self.manager.close.assert_not_awaited() self.probe.side_effect = RuntimeError(KEY) response = self.client.post("/api/admin/codex/provider/test", headers=NATIVE, json=SELECTION) self.assertEqual(response.status_code, 503) self.assertNotIn(KEY, response.text) def test_saved_model_check_uses_pinned_chat_key_and_caches_only_that_model(self): self.store.save(SELECTION) revision = self.store.revision() self.ns["STORE"].sessions["chat"] = {"codex_provider": "custom", "codex_provider_revision": revision} self.store.save({**SELECTION, "api_key": "new-synthetic-key"}) received = [] async def probe(selected): received.append(dict(selected)) return {**provider.test_result("ready"), "compatibility": "verified", "model": selected["model"]} self.probe.side_effect = probe response = self.client.post("/api/admin/codex/provider/test", headers=NATIVE, json={"model": "checked/model", "session_id": "chat", "credential_id": revision}) self.assertEqual(response.status_code, 200, response.text) self.assertEqual((received[0]["api_key"], received[0]["model"]), (KEY, "checked/model")) self.assertNotIn(KEY, response.text) cached = self.store.cached_catalog(received[0])["model_capabilities"] self.assertEqual(cached["checked/model"]["compatibility"], "verified") self.assertNotIn("checked/model", self.store.cached_catalog()["model_capabilities"]) response = self.client.post("/api/admin/codex/provider/test", headers=NATIVE, json={"model": "checked/model", "credential_id": revision}) self.assertEqual(response.status_code, 409) self.assertEqual(len(received), 1) async def test_cancelled_save_keeps_admission_until_disk_commit_finishes(self): entered, release = threading.Event(), threading.Event() save = self.store.save def delayed_save(selected): entered.set() if not release.wait(5): raise RuntimeError("test release timeout") save(selected) with patch.object(self.store, "save", side_effect=delayed_save): task = asyncio.create_task(self.ns["mutate_codex_provider"](dict(SELECTION))) self.assertTrue(await asyncio.to_thread(entered.wait, 3)) task.cancel() await asyncio.sleep(0) self.assertTrue(self.ns["CODEX_PROVIDER_SETTINGS_LOCK"].locked()) self.assertFalse(self.ns["CODEX_GOALS_RECONFIGURING"]) self.assertFalse(task.done()) release.set() with self.assertRaises(asyncio.CancelledError): await task self.assertFalse(self.ns["CODEX_GOALS_RECONFIGURING"]) self.assertTrue(self.store.status()["configured"]) class DiscoveryTests(unittest.TestCase): def test_native_effort_choices_require_declared_openai_responses_and_exact_model(self): model = "openai/openai/gpt-6-astra" native_models = {"models": [{"slug": "gpt-6-astra", "supported_reasoning_levels": [{"effort": "low"}, {"effort": "high"}, {"effort": "ultra"}]}]} metadata = {"owned_by": "openai", "mode": "responses"} capability = provider.discovered_model_capability(metadata, model, native_models) self.assertEqual(capability["reasoning_efforts"], ["low", "high", "ultra"]) self.assertEqual(capability["compatibility"], "unverified") self.assertIsNone(capability["reasoning_summary_supported"]) catalog = {"model_capabilities": {model: capability}} self.assertEqual(provider.runtime_effort({"model": model}, catalog, "ultra"), "ultra") for changes, model_id in (({}, "openai/gpt-6-astra-preview"), ({"owned_by": "other"}, model), ({"mode": "chat"}, model), ({"owned_by": None}, model), ({"mode": None}, model), ({"reasoning_supported": False}, model), ({"reasoning_supported": True}, model), ({"supports_reasoning": False}, model), ({"reasoning_efforts": []}, model), ({"supported_reasoning_efforts": None}, model), ({"capabilities": {"reasoning": False}}, model), ({"reasoning": {"efforts": []}}, model)): with self.subTest(changes=changes, model=model_id): result = provider.discovered_model_capability({**metadata, **changes}, model_id, native_models) self.assertEqual(result["reasoning_efforts"], []) explicit = provider.discovered_model_capability({**metadata, "supported_reasoning_levels": [{"effort": "medium"}, {"effort": "high"}, {"effort": "high"}, {"effort": "invalid"}]}, model, native_models) self.assertEqual(explicit["reasoning_efforts"], ["medium", "high"]) denied = provider.discovered_model_capability({**metadata, "reasoning_supported": False, "supported_reasoning_levels": [{"effort": "high"}]}, model, native_models) self.assertEqual(denied["reasoning_efforts"], []) def test_summary_capability_is_independent_and_requires_explicit_boolean_evidence(self): for model in ("gpt-6-astra", "unknown/provider-model"): for metadata, expected in (({}, None), ({"reasoning_efforts": ["high"]}, None), ({"supports_reasoning_summary_parameter": True}, True), ({"reasoning_summary_supported": False, "reasoning_efforts": ["high"]}, False), ({"capabilities": {"reasoning_summary": True}}, True), ({"reasoning": {"summary_supported": True}}, True), ({"reasoning_summary_supported": "true"}, None)): with self.subTest(model=model, metadata=metadata): capability = provider.discovered_model_capability(metadata, model) self.assertIs(capability["reasoning_summary_supported"], expected) selected = {**SELECTION, "model": model} summary = provider.runtime_summary(selected, {"model_capabilities": {model: capability}}) self.assertEqual(summary, "auto" if expected is True else "none") overrides = provider.turn_overrides(model, summary=summary) self.assertEqual(overrides["summary"], summary) self.assertNotIn("effort", overrides) self.assertIsNone(overrides["collaborationMode"]["settings"]["reasoning_effort"]) self.assertEqual(provider.native_config({**selected, "reasoning_summary": summary})["model_reasoning_summary"], summary) def test_owned_models_endpoint_success_auth_failure_and_redirect_no_follow(self): requests = [] class Handler(BaseHTTPRequestHandler): status = 200 def log_message(self, *args): pass def do_GET(self): requests.append((self.path, self.headers.get("Authorization"))) self.send_response(self.status) self.send_header("Location", "/must-not-follow") self.end_headers() self.wfile.write(json.dumps({"data": [ {"id": "model/one"}, {"id": "model/two", "type": "chat", "reasoning_efforts": ["low", "high", "not-a-native-effort"]}, {"id": "model/one"}, {"id": "invalid model"}, {"id": "text-embedding-3-small"}, {"id": "whisper-1"}, {"id": "provider/new-name", "task": "reranking"}, {"id": "provider/art", "architecture": {"output_modalities": ["image"]}}, {"id": "model/three", "reasoning_supported": False, "reasoning_efforts": ["high"]}, {"id": "chat-with-embedding-tools", "output_modalities": ["text", "image"]} ]}).encode()) server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: selected = {"base_url": f"http://127.0.0.1:{server.server_port}/v1", "api_key": KEY} ready = provider.discover_models(selected) self.assertEqual([model["value"] for model in ready["models"]], ["model/one", "model/two", "model/three", "chat-with-embedding-tools"]) self.assertEqual(ready["model_capabilities"]["model/one"], {"kind": "unknown", "compatibility": "unverified", "reasoning_efforts": [], "reasoning_supported": None, "reasoning_summary_supported": None}) self.assertEqual(ready["model_efforts"], {"model/one": [], "model/two": [{"value": "low", "label": "Low"}, {"value": "high", "label": "High"}], "model/three": [], "chat-with-embedding-tools": []}) self.assertFalse(ready["model_capabilities"]["model/three"]["reasoning_supported"]) self.assertEqual((ready["efforts"], ready["default_effort"]), ([], "")) self.assertTrue(ready["ok"]) Handler.status = 401 self.assertEqual(provider.discover_models(selected)["status"], "authentication_failed") Handler.status = 302 self.assertEqual(provider.discover_models(selected)["status"], "failed") self.assertEqual(requests, [("/v1/models", "Bearer " + KEY)] * 3) self.assertNotIn(KEY, json.dumps(ready)) finally: server.shutdown() server.server_close() thread.join() class ManagerGenerationTests(unittest.IsolatedAsyncioTestCase): async def test_simultaneous_generations_freeze_credentials_without_touching_normal_manager(self): with tempfile.TemporaryDirectory(prefix="provider-managers-") as temporary: store = provider.ProviderStore(Path(temporary) / "private") store.save(SELECTION) revision = store.revision() source = Path(__file__).with_name("agent_server.py") names = {"codex_app_server_managers", "existing_codex_app_server_manager", "codex_app_server_manager"} nodes = [node for node in ast.parse(source.read_text()).body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name in names] created = [] def factory(*args, **kwargs): manager = SimpleNamespace(client=SimpleNamespace(), add_notification_handler=Mock(), close=AsyncMock(), options=kwargs) created.append(manager) return manager normal_env = {"OPENAI_API_KEY": "normal-synthetic"} ns = {"asyncio": asyncio, "CODEX_PROVIDER_STORE": store, "codex_provider": provider, "CODEX_APP_SERVER_MANAGER": None, "CODEX_CUSTOM_APP_SERVER_MANAGERS": {}, "CODEX_APP_SERVER_MANAGER_EPOCH": 0, "CODEX_GOALS_CONFIG_LOCK": asyncio.Lock(), "CODEX_APP_SERVER_MANAGER_LOCK": asyncio.Lock(), "CODEX_GOALS_ENABLED": True, "ensure_provider_manager_factory_admission": lambda **kwargs: None, "CodexAppServerManager": factory, "CODEX_BIN": "unused", "existing_cwd": lambda value: value, "DEFAULT_CWD": temporary, "SERVER_VERSION": "test", "HTTPException": HTTPException, "codex_app_server_env": lambda selected=None: provider.native_environment(normal_env, selected) if selected else dict(normal_env)} for name in ("CODEX_APP_SERVER_TIMEOUT_SECONDS", "CODEX_APP_SERVER_LIFECYCLE_TIMEOUT_SECONDS", "CODEX_APP_SERVER_JSONL_LIMIT_BYTES", "CODEX_APP_SERVER_NOTIFICATION_QUEUE_LIMIT"): ns[name] = 10 for name in ("handle_codex_server_request", "register_codex_app_server_child", "unregister_codex_app_server_child", "project_codex_notification", "cache_codex_approval_item"): ns[name] = Mock() exec(compile(ast.fix_missing_locations(ast.Module(body=[ast.ImportFrom(module="__future__", names=[ast.alias(name="annotations")], level=0), *nodes], type_ignores=[])), str(source), "exec"), ns) normal = await ns["codex_app_server_manager"]() old_session = {"codex_provider": "custom", "codex_provider_revision": revision} old = await ns["codex_app_server_manager"](old_session) store.save({"base_url": SELECTION["base_url"], "api_key": "replacement-synthetic"}) new = await ns["codex_app_server_manager"]({"codex_provider": "custom", "model": "another-model"}) self.assertIs(await ns["codex_app_server_manager"](old_session), old) with patch.object(store, "for_session", side_effect=AssertionError("lookup must not read credentials")): self.assertIs(ns["existing_codex_app_server_manager"](old_session), old) self.assertIs(await ns["codex_app_server_manager"](), normal) self.assertIsNot(old, new) self.assertEqual(old.options["env_factory"]()[provider.ENV_KEY], KEY) self.assertEqual(new.options["env_factory"]()[provider.ENV_KEY], "replacement-synthetic") self.assertEqual(normal.options["env_factory"](), normal_env) self.assertNotIn("OPENAI_API_KEY", old.options["env_factory"]()) self.assertIn('cli_auth_credentials_store="ephemeral"', old.options["app_server_args"]) self.assertNotIn(KEY, str(old.options["app_server_args"])) self.assertEqual(ns["codex_app_server_managers"](), tuple(created)) for manager in created: manager.close.assert_not_awaited() class NativeCatalogTests(unittest.IsolatedAsyncioTestCase): async def test_owned_catalog_keeps_tool_metadata_and_never_uses_normal_credentials(self): original = {"slug": "builtin/model", "default_reasoning_level": "low", "supported_reasoning_levels": [{"effort": "low"}, {"effort": "high"}], "default_reasoning_summary": "detailed", "supports_experimental_context": True, "multi_agent_reasoning_effort": "high", "use_responses_lite": True, "model_messages": {"base_instructions": "native policy", "instructions_template": "native template"}, "tool_mode": "native-tools", "context_window": 272000} with tempfile.TemporaryDirectory(prefix="provider-native-catalog-") as temporary: target = Path(temporary) / "models.json" async def read(command, *, prompt, cwd, env, timeout): self.assertEqual(command[-2:], ["debug", "models"]) self.assertEqual((env["HOME"], env["CODEX_HOME"], env["TMPDIR"]), (cwd, cwd, cwd)) self.assertNotIn(KEY, str(command) + str(env)) self.assertNotIn("OPENAI_API_KEY", env) self.assertNotIn(provider.ENV_KEY, env) self.assertNotIn("AGENTSDOCK_AGENT_TOKEN", env) self.assertTrue(Path(cwd).is_dir()) self.assertEqual(timeout, 15) return json.dumps({"models": [original]}) with patch.object(provider, "run_isolated_command", side_effect=read): result = await provider.prepare_native_catalog("unused-native", { "PATH": "/synthetic/bin", "HOME": "/normal-home", "CODEX_HOME": "/normal-codex", "OPENAI_API_KEY": KEY, provider.ENV_KEY: KEY, "AGENTSDOCK_AGENT_TOKEN": KEY, }, target) self.assertEqual(result, str(target)) self.assertEqual(target.stat().st_mode & 0o777, 0o600) self.assertEqual(list(Path(temporary).iterdir()), [target]) model = json.loads(target.read_text())["models"][0] for key in ("model_messages", "tool_mode", "context_window", "supported_reasoning_levels"): self.assertEqual(model[key], original[key]) self.assertIsNone(model["default_reasoning_level"]) self.assertIsNone(model["multi_agent_reasoning_effort"]) self.assertEqual(model["default_reasoning_summary"], "none") self.assertTrue(model["supports_reasoning_summary_parameter"]) self.assertFalse(model["supports_experimental_context"]) self.assertFalse(model["use_responses_lite"]) self.assertEqual(original["default_reasoning_level"], "low") async def test_invalid_native_catalog_is_safe_and_preserves_previous_file(self): with tempfile.TemporaryDirectory(prefix="provider-native-catalog-") as temporary: target = Path(temporary) / "models.json" target.write_text("previous-catalog") for raw in (KEY, '{"models": []}', '{"models": [{"slug": "same"}, {"slug": "same"}]}'): with self.subTest(raw_is_json=raw != KEY), patch.object(provider, "run_isolated_command", AsyncMock(return_value=raw)): with self.assertRaises(HTTPException) as caught: await provider.prepare_native_catalog("unused-native", {}, target) self.assertEqual(caught.exception.status_code, 503) self.assertNotIn(KEY, str(caught.exception.detail)) self.assertEqual(target.read_text(), "previous-catalog") self.assertEqual(list(Path(temporary).iterdir()), [target]) class ProbeTests(unittest.IsolatedAsyncioTestCase): def setUp(self): self.prepare_catalog = AsyncMock(side_effect=lambda executable, env, path: str(path)) self.enterContext(patch.object(provider, "prepare_native_catalog", self.prepare_catalog)) async def test_shared_native_registration_preserves_effective_shell_exclusions(self): self.assertFalse(any(value.startswith("shell_environment_policy.exclude=") for value in provider.registration_args(SELECTION))) factory = FakeProcessFactory() client = CodexAppServerClient("unused", cwd="/tmp", env_factory=dict, process_factory=factory, protected_env_keys=(provider.ENV_KEY,), request_timeout=1) self.addAsyncCleanup(client.close) factory.process.responders["config/read"] = lambda _: {"config": {"shell_environment_policy": {"exclude": ["OPERATOR_SECRET_*", "PROJECT_PRIVATE"]}}} for method in ("thread/start", "thread/resume", "thread/fork"): factory.process.responders[method] = lambda _: {"thread": {"id": "synthetic"}} original = {"cwd": "/synthetic-project", "config": {"shell_environment_policy.exclude": ["THREAD_PRIVATE"], "unrelated": True}} await client.request(method, original) received = factory.process.messages[-1]["params"] self.assertEqual(received["config"]["shell_environment_policy.exclude"], ["OPERATOR_SECRET_*", "PROJECT_PRIVATE", "THREAD_PRIVATE", provider.ENV_KEY]) self.assertTrue(received["config"]["unrelated"]) self.assertEqual(original["config"]["shell_environment_policy.exclude"], ["THREAD_PRIVATE"]) self.assertEqual(factory.process.messages[-2]["params"]["cwd"], "/synthetic-project") factory.process.responders["config/read"] = lambda _: {"config": {"shell_environment_policy": {"exclude": "invalid"}}} before = len([message for message in factory.process.messages if message["method"] == "thread/start"]) with self.assertRaises(CodexAppServerProtocolError): await client.request("thread/start", {"cwd": "/synthetic-project"}) self.assertEqual(len([message for message in factory.process.messages if message["method"] == "thread/start"]), before) async def test_custom_upstream_error_notification_and_stderr_are_redacted(self): factory = FakeProcessFactory() client = CodexAppServerClient("unused", cwd="/tmp", env_factory=dict, sensitive_values=(KEY,), process_factory=factory, request_timeout=1) self.addAsyncCleanup(client.close) seen = [] client.add_notification_handler(seen.append) def reject(message): factory.process.feed_stderr(KEY) factory.process.feed({"method": "error", "params": {"message": KEY, "data": [KEY]}}) factory.process.feed({"id": message["id"], "error": {"message": KEY, "data": KEY}}) return NO_RESPONSE factory.process.responders["config/read"] = reject with self.assertRaises(CodexAppServerRequestError) as caught: await client.request("config/read", {}) await asyncio.sleep(0) self.assertNotIn(KEY, str(caught.exception)) self.assertNotIn(KEY, str(caught.exception.error)) self.assertNotIn(KEY, str(seen)) self.assertTrue(seen) self.assertEqual(client.stderr_tail, []) async def test_probe_native_tool_roundtrip_followup_and_isolated_cleanup(self): config = {**provider.native_config(SELECTION), "cli_auth_credentials_store": "ephemeral", "mcp_servers": {"inherited": {}}} native = SimpleNamespace(client=SimpleNamespace(), start=AsyncMock(), request=AsyncMock(return_value={"config": config}), start_thread=AsyncMock(return_value="ephemeral"), read_thread=AsyncMock(return_value={"ephemeral": True, "path": None}), close=AsyncMock()) captured, turns = {}, [] token = "" async def start_turn(thread, inputs, *, overrides): nonlocal token self.assertEqual((thread, overrides), ("ephemeral", {**provider.turn_overrides(SELECTION["model"], summary="auto" if len(turns) == 2 else "none"), "environments": []})) if not turns: reply = await captured["server_request_handler"](1, "item/tool/call", { "threadId": thread, "tool": provider.TEST_TOOL, "arguments": {}}) self.assertTrue(reply["success"]) token = reply["contentItems"][0]["text"] else: self.assertNotIn(token, str(inputs), "Follow-up must recover the previous result from native context") turn = SimpleNamespace(next_notification=AsyncMock(side_effect=[ {"method": "turn/plan/updated", "params": {"plan": [{"step": "Compatibility check", "status": "completed"}]}}, {"method": "item/completed", "params": {"item": {"type": "agentMessage", "text": token}}}, {"method": "turn/completed", "params": {"turn": {"status": "completed"}}}]), close=AsyncMock()) turns.append(turn) return turn native.start_turn = AsyncMock(side_effect=start_turn) def factory(*args, **kwargs): captured.update(kwargs) captured["environment"] = kwargs["env_factory"]().copy() return native verify = AsyncMock() result = await provider.test_connection(SELECTION, executable="synthetic-unused", environment={"OPENAI_API_KEY": "old", "HOME": "/synthetic-home", "CODEX_HOME": "/synthetic-existing-codex"}, manager_factory=factory, verify_protocol=verify) self.assertTrue(result["ok"]) self.assertEqual(result["compatibility"], "verified") self.assertTrue(all(result["checks"].values())) self.assertEqual(len(turns), 3) self.assertIsNone(result["reasoning_summary_supported"]) self.assertEqual(result["summary_check"], "inconclusive") for turn in turns: turn.close.assert_awaited_once() verify.assert_awaited_once() self.assertNotIn(KEY, str(captured["app_server_args"])) startup = tomllib.loads("\n".join(captured["app_server_args"][1::2])) self.assertEqual(startup["model_providers"][provider.PROVIDER_ID]["request_max_retries"], 0) self.assertEqual(startup["model_providers"][provider.PROVIDER_ID]["stream_max_retries"], 0) self.assertEqual(captured["environment"][provider.ENV_KEY], KEY) self.assertNotIn("OPENAI_API_KEY", captured["environment"]) self.assertEqual((captured["environment"]["HOME"], captured["environment"]["CODEX_HOME"]), ("/synthetic-home", captured["cwd"])) params = native.start_thread.call_args.args[0] self.assertEqual(params["config"]["model_providers"][provider.PROVIDER_ID]["request_max_retries"], 0) self.assertEqual(params["config"]["model_providers"][provider.PROVIDER_ID]["stream_max_retries"], 0) self.prepare_catalog.assert_awaited_once() self.assertEqual(params["config"]["model_catalog_json"], str(Path(captured["cwd"]) / "models.json")) self.assertEqual(params["environments"], []) self.assertEqual(params["dynamicTools"][0]["name"], provider.TEST_TOOL) self.assertTrue(params["config"]["tools.update_plan.enabled"]) self.assertEqual(params["config"]["mcp_servers"], {"inherited": {"enabled": False}}) self.assertEqual(params["approvalPolicy"], "never") self.assertFalse(Path(captured["cwd"]).exists()) native.close.assert_awaited_once() self.assertEqual(captured["env_factory"](), {}) async def test_text_only_reply_does_not_claim_tool_compatibility(self): turn = SimpleNamespace(next_notification=AsyncMock(side_effect=[ {"method": "item/completed", "params": {"item": {"type": "agentMessage", "text": "CONNECTION_OK"}}}, {"method": "turn/completed", "params": {"turn": {"status": "completed"}}}]), close=AsyncMock()) config = {**provider.native_config(SELECTION), "cli_auth_credentials_store": "ephemeral"} native = SimpleNamespace(client=SimpleNamespace(), start=AsyncMock(), request=AsyncMock(return_value={"config": config}), start_thread=AsyncMock(return_value="ephemeral"), read_thread=AsyncMock(return_value={"ephemeral": True, "path": None}), start_turn=AsyncMock(return_value=turn), close=AsyncMock()) result = await provider.test_connection(SELECTION, executable="unused", environment={}, manager_factory=lambda *args, **kwargs: native, verify_protocol=AsyncMock()) self.assertEqual((result["ok"], result["compatibility"], result["status"]), (False, "unverified", "inconclusive")) native.start_turn.assert_awaited_once() native.close.assert_awaited_once() async def test_optional_summary_check_requires_visible_evidence_and_preserves_basic_success(self): cases = [ ([{"method": "item/reasoning/summaryTextDelta", "params": {"delta": "Compare the products."}}], True), ([{"method": "item/completed", "params": {"item": {"type": "reasoning", "summary": ["Compare the products."]}}}], True), ([{"method": "item/reasoning/textDelta", "params": {"delta": "Raw text is not summary support."}}], None), ([], None), ([{"method": "error", "params": {"error": {"message": "Unsupported parameter " + KEY, "param": "reasoning.summary"}}}], False), ([{"method": "error", "params": {"error": "Unsupported parameter reasoning.effort " + KEY}}], None), ([{"method": "error", "params": {"error": "401 unauthorized " + KEY}}], None), ([TimeoutError()], None), ] for model in ("gpt-6-astra", "unknown/provider-model"): for notifications, expected in cases: with self.subTest(model=model, summary_supported=expected, notifications=len(notifications)): selected = {**SELECTION, "model": model} native = SimpleNamespace(client=SimpleNamespace(), start=AsyncMock(), close=AsyncMock(), request=AsyncMock(return_value={"config": {**provider.native_config(selected), "cli_auth_credentials_store": "ephemeral"}}), start_thread=AsyncMock(return_value="ephemeral"), read_thread=AsyncMock(return_value={"ephemeral": True, "path": None})) captured, turns = {}, [] token = "" async def start_turn(thread, inputs, *, overrides): nonlocal token if not turns: reply = await captured["server_request_handler"](1, "item/tool/call", { "threadId": thread, "tool": provider.TEST_TOOL, "arguments": {}}) token = reply["contentItems"][0]["text"] optional = len(turns) == 2 self.assertEqual(overrides["summary"], "auto" if optional else "none") packets = notifications if optional else [ {"method": "turn/plan/updated", "params": {}}, {"method": "item/completed", "params": {"item": {"type": "agentMessage", "text": token}}}] turn = SimpleNamespace(next_notification=AsyncMock(side_effect=[*packets, {"method": "turn/completed", "params": {"turn": {"status": "completed"}}}]), close=AsyncMock()) turns.append(turn) return turn native.start_turn = AsyncMock(side_effect=start_turn) def factory(*args, **kwargs): captured.update(kwargs) return native result = await provider.test_connection(selected, executable="unused", environment={}, manager_factory=factory, verify_protocol=AsyncMock()) self.assertEqual((result["ok"], result["compatibility"]), (True, "verified")) self.assertTrue(all(result["checks"].values())) self.assertIs(result["reasoning_summary_supported"], expected) self.assertEqual(result["summary_check"], "supported" if expected is True else "unsupported" if expected is False else "inconclusive") self.assertNotIn(KEY, str(result)) self.assertEqual(native.start_turn.await_count, 3) for turn in turns: turn.close.assert_awaited_once() native.close.assert_awaited_once() async def test_protocol_or_native_errors_are_safe_and_never_retry(self): verify = AsyncMock(side_effect=RuntimeError(KEY)) factory = AsyncMock() result = await provider.test_connection(SELECTION, executable="unused", environment={}, manager_factory=factory, verify_protocol=verify) self.assertEqual(result["status"], "failed") self.assertNotIn(KEY, str(result)) factory.assert_not_called() async def test_first_native_error_finishes_before_completion_and_closes_without_retry(self): cases = [ (True, "Connection failed: error sending request " + KEY, "connection_failed"), (False, "Upstream returned 401 Unauthorized " + KEY, "authentication_failed"), (True, "Unknown failure " + KEY, "failed"), (False, "Unsupported parameter: reasoning.effort " + KEY, "unsupported_parameter"), (False, {"message": "Invalid value: custom. Supported values: function", "param": "tools[0].type", "private": KEY}, "unsupported_parameter"), ] for will_retry, details, expected in cases: with self.subTest(will_retry=will_retry, expected=expected): turn = SimpleNamespace(next_notification=AsyncMock(side_effect=[ {"method": "error", "params": {"threadId": "ephemeral", "turnId": "probe-turn", "willRetry": will_retry, "error": {"message": "Reconnecting... 1/4", "additionalDetails": details}}}, {"method": "turn/completed", "params": {"turn": {"status": "completed"}}}, ]), close=AsyncMock()) config = {**provider.native_config(SELECTION), "cli_auth_credentials_store": "ephemeral"} native = SimpleNamespace(client=SimpleNamespace(), start=AsyncMock(), request=AsyncMock(return_value={"config": config}), start_thread=AsyncMock(return_value="ephemeral"), read_thread=AsyncMock(return_value={"ephemeral": True, "path": None}), start_turn=AsyncMock(return_value=turn), close=AsyncMock()) result = await asyncio.wait_for(provider.test_connection(SELECTION, executable="unused", environment={}, manager_factory=lambda *args, **kwargs: native, verify_protocol=AsyncMock()), 1) self.assertEqual(result["status"], expected) self.assertNotIn(KEY, str(result)) turn.next_notification.assert_awaited_once() turn.close.assert_awaited_once() native.start_turn.assert_awaited_once() native.close.assert_awaited_once() async def test_cancel_during_spawn_joins_owned_start_and_close(self): entered, release = asyncio.Event(), asyncio.Event() async def start(): entered.set() await release.wait() native = SimpleNamespace(client=SimpleNamespace(), start=start, close=AsyncMock(), request=AsyncMock()) task = asyncio.create_task(provider.test_connection(SELECTION, executable="unused", environment={}, manager_factory=lambda *args, **kwargs: native, verify_protocol=AsyncMock())) await entered.wait() task.cancel() await asyncio.sleep(0) self.assertFalse(task.done()) native.close.assert_not_awaited() release.set() with self.assertRaises(asyncio.CancelledError): await task native.close.assert_awaited_once() native.request.assert_not_awaited() if __name__ == "__main__": unittest.main()