tests/test_advisor_config.py

"""Behavior tests for the local Advisor configuration foundation."""

from __future__ import annotations

import importlib.util
import io
import json
import os
import signal
import stat
import subprocess
import sys
import tempfile
import threading
import time
import unittest
from contextlib import redirect_stderr
from pathlib import Path
from unittest import mock


MODULE = Path(__file__).resolve().parents[1] / "scripts" / "advisor_config.py"
SPEC = importlib.util.spec_from_file_location("advisor_config", MODULE)
assert SPEC and SPEC.loader
config = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = config
SPEC.loader.exec_module(config)

CONFIG_FACADE_NAMES = (
    "AppServerDiscovery",
    "ConfigError",
    "DiscoveryError",
    "DiscoveryUnavailable",
    "MAX_LIVE_CONFIG_BYTES",
    "ProcessUnavailable",
    "RevisionConflict",
    "_atomic_write",
    "_run_bounded_output",
    "_run_compatibility_wrapper",
    "add_manual_candidate",
    "baseline_settings",
    "clear_usage_journal",
    "compatibility_is_current",
    "live_config_path",
    "load_catalog",
    "load_live_config",
    "load_settings",
    "load_shipped_models",
    "normalize_discovery_pages",
    "read_json",
    "record_compatibility",
    "refresh_catalog",
    "resolve_selection",
    "restore_prior_settings",
    "save_preset",
    "save_settings",
    "set_deadline",
    "set_usage_journal",
    "shipped_default_launch_eligible",
    "state_paths",
    "validate_compatibility",
    "validate_journal_record",
    "validate_settings",
    "validate_shipped_models",
    "write_usage_journal",
)


class AdvisorConfigTests(unittest.TestCase):
    def setUp(self) -> None:
        self.tmp = tempfile.TemporaryDirectory()
        self.codex_home = Path(self.tmp.name) / "codex"
        self.codex_home.mkdir(mode=0o700)
        self.paths = config.state_paths(self.codex_home)

    def tearDown(self) -> None:
        self.tmp.cleanup()

    def test_config_facade_exposes_plugin_test_namespace(self) -> None:
        for name in CONFIG_FACADE_NAMES:
            with self.subTest(name=name):
                self.assertTrue(hasattr(config, name), name)

    def test_plugin_file_anchors_remain_under_root(self) -> None:
        root = MODULE.parents[1]
        self.assertEqual(config.live_config_path(), root / "advisor.toml")
        self.assertEqual(config.load_live_config()["path"], str(root / "advisor.toml"))
        models = root / "models.json"
        self.assertEqual(config.load_shipped_models(), json.loads(models.read_text()))
        for name in ("run-advisor.sh", "inspect-parent-runtime.sh"):
            with self.subTest(name=name):
                self.assertEqual((MODULE.parent / name).resolve().parent, root / "scripts")
                self.assertTrue((MODULE.parent / name).is_file())

    def test_defaults_are_read_only_and_do_not_initialize_state(self) -> None:
        self.assertEqual(
            config.load_settings(self.paths)["selections"]["standard"],
            {"model": "gpt-6.1-terra", "effort": "high"},
        )
        self.assertFalse(self.paths.root.exists())

    def test_invalid_json_duplicate_and_unknown_settings_fail_closed(self) -> None:
        self.paths.root.mkdir(mode=0o700)
        self.paths.settings.write_text(
            '{"schema_version":1,"schema_version":1}', encoding="utf-8"
        )
        os.chmod(self.paths.settings, 0o600)
        with self.assertRaises(config.ConfigError):
            config.load_settings(self.paths)
        bad = config.baseline_settings()
        bad["unexpected"] = True
        with self.assertRaises(config.ConfigError):
            config.validate_settings(bad)

    def test_json_null_and_boolean_schema_version_do_not_become_defaults(self) -> None:
        self.paths.root.mkdir(mode=0o700)
        self.paths.settings.write_text("null\n", encoding="utf-8")
        os.chmod(self.paths.settings, 0o600)
        with self.assertRaises(config.ConfigError):
            config.load_settings(self.paths)
        boolean = config.baseline_settings()
        boolean["schema_version"] = True
        with self.assertRaises(config.ConfigError):
            config.validate_settings(boolean)

    def test_symlink_and_permissive_paths_are_rejected(self) -> None:
        self.paths.root.mkdir(mode=0o700)
        target = Path(self.tmp.name) / "target.json"
        target.write_text("{}", encoding="utf-8")
        os.chmod(target, 0o600)
        self.paths.settings.symlink_to(target)
        with self.assertRaises(config.ConfigError):
            config.load_settings(self.paths)
        self.paths.settings.unlink()
        os.chmod(self.paths.root, 0o755)
        with self.assertRaises(config.ConfigError):
            config.load_catalog(self.paths)

    def test_revision_guards_concurrent_writers(self) -> None:
        initial = config.baseline_settings()
        first = config.save_settings(initial, expected_revision=0, paths=self.paths)
        self.assertEqual(first["revision"], 1)
        outcomes: list[object] = []

        def writer() -> None:
            try:
                outcomes.append(
                    config.save_settings(first, expected_revision=1, paths=self.paths)[
                        "revision"
                    ]
                )
            except Exception as exc:  # expected for one writer
                outcomes.append(exc)

        threads = [threading.Thread(target=writer) for _ in range(2)]
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join()
        self.assertEqual(sum(item == 2 for item in outcomes), 1)
        self.assertEqual(
            sum(isinstance(item, config.RevisionConflict) for item in outcomes), 1
        )
        self.assertEqual(stat.S_IMODE(self.paths.root.stat().st_mode), 0o700)
        self.assertEqual(stat.S_IMODE(self.paths.settings.stat().st_mode), 0o600)

    def test_future_selector_is_a_candidate_not_a_tier_or_compatibility(self) -> None:
        future = "future/provider-model_2049:beta"
        catalog = config.add_manual_candidate(future, paths=self.paths)
        item = catalog["candidates"][0]
        self.assertEqual(item["model"], future)
        self.assertEqual(item["provenance"], "manual")
        self.assertEqual(catalog["compatibility"], [])

    def test_compatibility_binds_pair_and_transport_not_cli_provenance(self) -> None:
        version = "codex-cli 0.153.2"
        receipt = config.validate_compatibility(
            {
                "model": "future/x",
                "effort": "high",
                "codex_version": version,
                "transport_contract_version": "1.4",
                "verified_at": "1",
            }
        )
        self.assertTrue(
            config.compatibility_is_current(
                receipt, model="future/x", effort="high"
            )
        )
        self.assertFalse(
            config.compatibility_is_current(
                receipt, model="future/x", effort="medium"
            )
        )
        self.assertTrue(
            config.shipped_default_launch_eligible(
                tier="standard", model="gpt-6.1-terra", effort="high"
            )
        )
        self.assertFalse(
            config.shipped_default_launch_eligible(
                tier="standard", model="gpt-6-astra", effort="high"
            )
        )

    def test_pristine_defaults_resolve_without_cli_probe_or_state(self) -> None:
        fake_bin = Path(self.tmp.name) / "bin"
        fake_bin.mkdir()
        fake = fake_bin / "codex"
        marker = Path(self.tmp.name) / "codex-probed"
        fake.write_text(
            '#!/bin/sh\nprintf called > "$CODEX_PROBE_MARKER"\n'
            'printf "codex-cli 0.153.2\\n"\n'
        )
        fake.chmod(0o700)
        with mock.patch.dict(
            os.environ,
            {"PATH": f"{fake_bin}:{os.environ['PATH']}", "CODEX_PROBE_MARKER": str(marker)},
        ):
            standard = config.resolve_selection("standard", paths=self.paths)
            self.assertFalse(marker.exists())
            specialist = config.resolve_selection("specialist", paths=self.paths)
            self.assertFalse(marker.exists())
        self.assertEqual(
            (standard["model"], standard["effort"]), ("gpt-6.1-terra", "high")
        )
        self.assertEqual(
            (specialist["model"], specialist["effort"]), ("gpt-6.1-sol", "high")
        )
        self.assertFalse(self.paths.root.exists())

    def test_bounded_probe_caps_output_and_reaps_on_timeout(self) -> None:
        probe = (
            "import os, sys, time; "
            "open(sys.argv[1], 'w').write(str(os.getpid())); "
            "os.write(1, b'x' * 128) if sys.argv[2] == 'output' else None; "
            "time.sleep(30)"
        )
        for case, limit, message in (
            ("output", 16, "local process probe exceeded output limit"),
            ("timeout", 16, "local process probe is unavailable"),
        ):
            with self.subTest(case=case):
                marker = Path(self.tmp.name) / f"{case}-pid"
                with self.assertRaisesRegex(config.ConfigError, message):
                    config._run_bounded_output(
                        [sys.executable, "-u", "-c", probe, str(marker), case],
                        timeout_seconds=0.5,
                        max_bytes=limit,
                    )
                pid = int(marker.read_text())
                with self.assertRaises(ProcessLookupError):
                    os.kill(pid, 0)

    def test_shipped_catalog_rejects_malformed_defaults_and_baselines(self) -> None:
        shipped = config.load_shipped_models()
        malformed = json.loads(json.dumps(shipped))
        malformed["defaults"]["standard"]["model"] = "gpt-6-astra"
        with self.assertRaises(config.ConfigError):
            config.validate_shipped_models(malformed)
        malformed = json.loads(json.dumps(shipped))
        malformed["models"].append(dict(malformed["models"][0]))
        with self.assertRaises(config.ConfigError):
            config.validate_shipped_models(malformed)
        malformed = json.loads(json.dumps(shipped))
        malformed["models"][0]["compatibility_baseline"]["efforts"] = [["high"]]
        with self.assertRaises(config.ConfigError):
            config.validate_shipped_models(malformed)

    def test_explicit_routes_fail_closed_when_transport_evidence_stales(self) -> None:
        pair = {"model": "future/model", "effort": "high"}
        config.add_manual_candidate(pair["model"], paths=self.paths)
        config.record_compatibility(
            pair, codex_version="codex-cli 0.153.2", paths=self.paths
        )
        stale = config.baseline_settings()
        stale["selections"]["standard"] = pair
        config.save_settings(stale, expected_revision=0, paths=self.paths)
        config.save_preset("future", pair, paths=self.paths)
        catalog = config.load_catalog(self.paths)
        catalog["compatibility"][0]["transport_contract_version"] = "1.3"
        self.paths.catalog.write_text(json.dumps(catalog) + "\n", encoding="utf-8")
        os.chmod(self.paths.catalog, 0o600)
        resolved = config.resolve_selection("standard", paths=self.paths)
        self.assertEqual(
            resolved["model"], "gpt-6.1-terra",
            "normal tiers must ignore stale saved selections",
        )
        with self.assertRaises(config.ConfigError):
            config.resolve_selection("standard", preset="future", paths=self.paths)

    def test_live_config_requires_exact_safe_toml_and_accepts_future_selector(self) -> None:
        candidate = Path(self.tmp.name) / "advisor.toml"
        with mock.patch.object(sys.modules["advisor_settings"], "live_config_path", return_value=candidate):
            candidate.write_text(
                "[standard]\nmodel = \"future/provider-2049\"\neffort = \"max\"\n"
                "[specialist]\nmodel = \"gpt-6-astra\"\neffort = \"high\"\n"
            )
            live = config.load_live_config()
            self.assertEqual(live["pairs"]["standard"], {"model": "future/provider-2049", "effort": "max"})
            self.assertTrue(live["source_revision"].startswith("sha256:"))
            candidate.write_text("[standard]\nmodel = \"a\"\neffort = \"high\"\n")
            with self.assertRaisesRegex(config.ConfigError, "advisor.toml"):
                config.load_live_config()
            candidate.write_text(
                "[standard]\nmodel = \"a\"\neffort = \"bad\"\n"
                "[specialist]\nmodel = \"b\"\neffort = \"high\"\n"
            )
            with self.assertRaisesRegex(config.ConfigError, "advisor.toml"):
                config.load_live_config()
            candidate.unlink()
            candidate.symlink_to(Path(self.tmp.name) / "target")
            with self.assertRaisesRegex(config.ConfigError, "advisor.toml"):
                config.load_live_config()

    def test_live_config_opens_nofollow_and_rejects_nonregular_files_without_blocking(self) -> None:
        original_open = os.open
        candidate = Path(self.tmp.name) / "advisor.toml"
        target = Path(self.tmp.name) / "target.toml"
        target.write_text(
            '[standard]\nmodel = "gpt-6.1-terra"\neffort = "high"\n'
            '[specialist]\nmodel = "gpt-6.1-sol"\neffort = "high"\n',
            encoding="utf-8",
        )
        candidate.write_bytes(target.read_bytes())

        def replace_before_open(path: Path, flags: int) -> int:
            candidate.unlink()
            candidate.symlink_to(target)
            return original_open(path, flags)

        with mock.patch.object(sys.modules["advisor_settings"], "live_config_path", return_value=candidate):
            with mock.patch.object(os, "open", side_effect=replace_before_open):
                with self.assertRaisesRegex(config.ConfigError, "advisor.toml"):
                    config.load_live_config()
            candidate.unlink()
            os.mkfifo(candidate)
            started = time.monotonic()
            with self.assertRaisesRegex(config.ConfigError, "regular file"):
                config.load_live_config()
            self.assertLess(time.monotonic() - started, 1.0)

    def test_live_config_caps_a_file_that_grows_after_open(self) -> None:
        candidate = Path(self.tmp.name) / "advisor.toml"
        candidate.write_text(
            '[standard]\nmodel = "gpt-6.1-terra"\neffort = "high"\n'
            '[specialist]\nmodel = "gpt-6.1-sol"\neffort = "high"\n',
            encoding="utf-8",
        )
        delivered = 0

        def growing_read(descriptor: int, count: int) -> bytes:
            nonlocal delivered
            self.assertLessEqual(count, config.MAX_LIVE_CONFIG_BYTES + 1 - delivered)
            chunk = b"x" * count
            delivered += len(chunk)
            return chunk

        with (
            mock.patch.object(sys.modules["advisor_settings"], "live_config_path", return_value=candidate),
            mock.patch.object(os, "read", side_effect=growing_read),
        ):
            with self.assertRaisesRegex(config.ConfigError, "too large"):
                config.load_live_config()
        self.assertEqual(delivered, config.MAX_LIVE_CONFIG_BYTES + 1)

    def test_settings_mutations_keep_one_explicit_prior_revision(self) -> None:
        initial = config.save_settings(
            config.baseline_settings(), expected_revision=0, paths=self.paths
        )
        updated = config.set_deadline(120, paths=self.paths)
        self.assertEqual(updated["deadline_seconds"], 120)
        prior = config.validate_settings(config.read_json(self.paths.prior_settings))
        self.assertEqual(prior["revision"], initial["revision"])
        self.assertEqual(prior["deadline_seconds"], 300)
        restored = config.restore_prior_settings(paths=self.paths)
        self.assertEqual(restored["deadline_seconds"], 300)
        self.assertGreater(restored["revision"], updated["revision"])

    def test_failed_settings_write_preserves_active_and_distinct_prior_bytes(
        self,
    ) -> None:
        initial = config.save_settings(
            config.baseline_settings(), expected_revision=0, paths=self.paths
        )
        config.set_deadline(120, paths=self.paths)
        active_bytes = self.paths.settings.read_bytes()
        prior_bytes = self.paths.prior_settings.read_bytes()
        self.assertNotEqual(active_bytes, prior_bytes)
        original = sys.modules["advisor_settings"]._atomic_write

        def fail_settings(path: Path, value: object) -> None:
            if path == self.paths.settings:
                raise OSError("simulated write failure")
            original(path, value)

        with mock.patch.object(sys.modules["advisor_settings"], "_atomic_write", side_effect=fail_settings):
            with self.assertRaises(OSError):
                config.set_deadline(120, paths=self.paths)
        self.assertEqual(self.paths.settings.read_bytes(), active_bytes)
        self.assertEqual(self.paths.prior_settings.read_bytes(), prior_bytes)
        self.assertFalse(list(self.paths.root.glob(".*.tmp")))

    def test_explicit_restore_recovers_invalid_active_and_preserves_its_bytes(
        self,
    ) -> None:
        initial = config.save_settings(
            config.baseline_settings(), expected_revision=0, paths=self.paths
        )
        config.set_deadline(120, paths=self.paths)
        malformed = b'{"schema_version":1,"nested":' + b"[" * 1000
        self.paths.settings.write_bytes(malformed)
        os.chmod(self.paths.settings, 0o600)
        restored = config.restore_prior_settings(paths=self.paths)
        self.assertEqual(restored["deadline_seconds"], initial["deadline_seconds"])
        self.assertEqual(restored["revision"], initial["revision"] + 1)
        self.assertEqual(self.paths.invalid_settings.read_bytes(), malformed)
        self.assertEqual(stat.S_IMODE(self.paths.invalid_settings.stat().st_mode), 0o600)

    def test_restore_refuses_invalid_or_unsafe_backup(self) -> None:
        config.save_settings(config.baseline_settings(), expected_revision=0, paths=self.paths)
        self.paths.settings.write_text("null\n", encoding="utf-8")
        self.paths.prior_settings.write_text("null\n", encoding="utf-8")
        with self.assertRaises(config.ConfigError):
            config.restore_prior_settings(paths=self.paths)
        self.paths.prior_settings.unlink()
        target = Path(self.tmp.name) / "prior-target"
        target.write_text("{}\n", encoding="utf-8")
        self.paths.prior_settings.symlink_to(target)
        with self.assertRaises(config.ConfigError):
            config.restore_prior_settings(paths=self.paths)

    def test_restore_refuses_to_downgrade_a_newer_settings_schema(self) -> None:
        config.save_settings(config.baseline_settings(), expected_revision=0, paths=self.paths)
        prior_bytes = self.paths.prior_settings.read_bytes()
        future = {"schema_version": 2, "revision": 9, "future": {"setting": True}}
        future_bytes = (json.dumps(future) + "\n").encode()
        self.paths.settings.write_bytes(future_bytes)
        os.chmod(self.paths.settings, 0o600)
        with self.assertRaisesRegex(config.ConfigError, "refuses to downgrade"):
            config.restore_prior_settings(paths=self.paths)
        self.assertEqual(self.paths.settings.read_bytes(), future_bytes)
        self.assertEqual(self.paths.prior_settings.read_bytes(), prior_bytes)
        self.assertFalse(self.paths.invalid_settings.exists())

    def test_canary_supervisor_cancellation_reaps_owned_group_only(self) -> None:
        wrapper = Path(self.tmp.name) / "canary-wrapper.sh"
        child_marker = Path(self.tmp.name) / "canary-child"
        terminated = Path(self.tmp.name) / "canary-terminated"
        wrapper.write_text(
            "#!/bin/sh\n"
            "trap 'printf terminated > \"$2\"; exit 143' TERM\n"
            "sleep 20 &\n"
            "printf '%s' \"$!\" > \"$1\"\n"
            "wait\n",
            encoding="utf-8",
        )
        wrapper.chmod(0o700)
        unrelated = subprocess.Popen(["sleep", "20"])
        timer = threading.Timer(1.0, os.kill, args=(os.getpid(), signal.SIGTERM))
        try:
            timer.start()
            with self.assertRaises(config.ConfigError):
                config._run_compatibility_wrapper(
                    [str(wrapper), str(child_marker), str(terminated)],
                    deadline_seconds=30,
                    on_status=None,
                )
            self.assertTrue(child_marker.exists())
            self.assertTrue(terminated.exists())
            self.assertIsNone(unrelated.poll())
        finally:
            timer.cancel()
            unrelated.terminate()
            unrelated.wait(timeout=3)

    def test_paginated_discovery_uses_model_not_id_and_keeps_hidden_rows(self) -> None:
        model = lambda ident, dispatch, hidden=False: {
            "id": ident,
            "model": dispatch,
            "displayName": ident,
            "description": "model",
            "hidden": hidden,
            "isDefault": False,
            "defaultReasoningEffort": "high",
            "supportedReasoningEfforts": [
                {"reasoningEffort": "high", "description": ""}
            ],
        }
        pages = [
            {"data": [model("marketing-name", "dispatch/one")], "nextCursor": "next"},
            {"data": [model("hidden-name", "dispatch/two", True)], "nextCursor": None},
        ]
        rows = config.normalize_discovery_pages(pages, now=lambda: 42)
        self.assertEqual(rows[0]["id"], "marketing-name")
        self.assertEqual(rows[0]["model"], "dispatch/one")
        self.assertFalse(rows[1]["visible"])
        self.assertEqual(rows[1]["model"], "dispatch/two")

    def test_discovery_requires_schema_fields_and_ignores_optional_metadata(
        self,
    ) -> None:
        model = {
            "id": "friendly",
            "model": "dispatch/model",
            "displayName": "Friendly",
            "description": "model",
            "hidden": False,
            "isDefault": True,
            "defaultReasoningEffort": "high",
            "supportedReasoningEfforts": [
                {"reasoningEffort": "high", "description": "default", "extra": 1}
            ],
            "serviceTiers": [],
        }
        rows = config.normalize_discovery_pages(
            [{"data": [model], "nextCursor": None, "optional": True}]
        )
        self.assertEqual(rows[0]["model"], "dispatch/model")
        del model["description"]
        with self.assertRaises(config.DiscoveryError):
            config.normalize_discovery_pages([{"data": [model], "nextCursor": None}])

    def test_models_lists_shipped_catalog_and_local_candidates(self) -> None:
        config.add_manual_candidate("future/model", paths=self.paths)
        env = {**os.environ, "CODEX_HOME": str(self.codex_home)}
        result = subprocess.run(
            [str(MODULE.parent / "advisor-config.sh"), "--json", "models"],
            env=env,
            text=True,
            capture_output=True,
            check=False,
        )
        payload = json.loads(result.stdout)
        self.assertEqual(result.returncode, 0)
        self.assertIn(
            "gpt-6-astra", [item["model"] for item in payload["shipped"]["models"]]
        )
        self.assertEqual(payload["catalog"]["candidates"][0]["model"], "future/model")

    def test_discovery_rejects_loop_and_malformed_pages(self) -> None:
        with self.assertRaises(config.DiscoveryError):
            config.normalize_discovery_pages(
                [
                    {"data": [], "nextCursor": "again"},
                    {"data": [], "nextCursor": "again"},
                ]
            )
        with self.assertRaises(config.DiscoveryError):
            config.normalize_discovery_pages(
                [{"data": [{"id": "only-id"}], "nextCursor": None}]
            )

    def test_refresh_preserves_manual_and_marks_missing_discovery_hidden(self) -> None:
        config.add_manual_candidate("manual/future", paths=self.paths)
        model = {
            "id": "old",
            "model": "server/old",
            "displayName": "Old",
            "description": "model",
            "hidden": False,
            "isDefault": False,
            "defaultReasoningEffort": "high",
            "supportedReasoningEfforts": [
                {"reasoningEffort": "high", "description": ""}
            ],
        }
        config.refresh_catalog(
            lambda: [{"data": [model], "nextCursor": None}],
            paths=self.paths,
            now=lambda: 10,
        )
        updated = config.refresh_catalog(
            lambda: [{"data": [], "nextCursor": None}], paths=self.paths, now=lambda: 20
        )
        self.assertEqual(updated["candidates"][0]["model"], "manual/future")
        old = next(row for row in updated["candidates"] if row["model"] == "server/old")
        self.assertFalse(old["visible"])

    def test_offline_adapter_is_closed_and_does_not_mutate_catalog(self) -> None:
        before = config.load_catalog(self.paths)
        with self.assertRaises(config.DiscoveryUnavailable):
            config.AppServerDiscovery(
                lambda: (_ for _ in ()).throw(config.ProcessUnavailable("offline"))
            ).pages()
        self.assertEqual(config.load_catalog(self.paths), before)

    def test_adapter_paginates_and_closes_its_owned_connection(self) -> None:
        class Connection:
            def __init__(self) -> None:
                self.calls, self.closed = [], False

            def request(self, method, payload):
                self.calls.append((method, payload))
                if method == "initialize":
                    return {"serverInfo": {}}
                model = lambda ident: {
                    "id": ident,
                    "model": ident,
                    "displayName": ident,
                    "description": "model",
                    "hidden": False,
                    "isDefault": False,
                    "defaultReasoningEffort": "high",
                    "supportedReasoningEfforts": [
                        {"reasoningEffort": "high", "description": ""}
                    ],
                }
                return (
                    {"data": [model("a")], "nextCursor": "c"}
                    if "cursor" not in payload
                    else {"data": [model("b")], "nextCursor": None}
                )

            def notify(self, method, payload):
                self.calls.append((method, payload))

            def close(self):
                self.closed = True

        connection = Connection()
        pages = config.AppServerDiscovery(lambda: connection).pages()
        self.assertEqual(len(pages), 2)
        self.assertTrue(connection.closed)
        self.assertEqual(connection.calls[0][0], "initialize")
        self.assertEqual(connection.calls[0][1]["clientInfo"]["name"], "codex-advisor")
        self.assertEqual(connection.calls[1][0], "initialized")

    def test_cli_refresh_emits_progress_and_never_invokes_inference(self) -> None:
        fake_bin = Path(self.tmp.name) / "bin"
        fake_bin.mkdir()
        fake = fake_bin / "codex"
        fake.write_text("#!/bin/sh\nexit 1\n", encoding="utf-8")
        fake.chmod(0o700)
        env = {
            **os.environ,
            "CODEX_HOME": str(self.codex_home),
            "PATH": f"{fake_bin}:{os.environ['PATH']}",
        }
        result = subprocess.run(
            [str(MODULE.parent / "advisor-config.sh"), "--json", "models", "refresh"],
            env=env,
            text=True,
            capture_output=True,
            check=False,
        )
        self.assertEqual(result.returncode, 2)
        self.assertIn("discovery started", result.stderr)
        self.assertEqual(json.loads(result.stdout)["status"], "unavailable")
        self.assertFalse(
            self.paths.root.exists(), "a failed read-only refresh must not create state"
        )


if __name__ == "__main__":
    unittest.main()