diff --git a/.github/workflows/package-smoke.yml b/.github/workflows/package-smoke.yml index e15e28263d..042e596550 100644 --- a/.github/workflows/package-smoke.yml +++ b/.github/workflows/package-smoke.yml @@ -42,6 +42,7 @@ jobs: - name: Validate recovery and existing operator contracts run: | python -m compileall -q packages/loopx-codex-provider-routing/src + python packages/loopx-codex-provider-routing/smoke/native_host_smoke.py python packages/loopx-codex-provider-routing/smoke/codex_provider_routing_smoke.py python packages/loopx-codex-provider-routing/smoke/recovery_contracts_smoke.py python packages/loopx-codex-provider-routing/smoke/operator_smoke.py diff --git a/packages/loopx-codex-provider-routing/NATIVE.md b/packages/loopx-codex-provider-routing/NATIVE.md index 004da6144f..7ec88789e1 100644 --- a/packages/loopx-codex-provider-routing/NATIVE.md +++ b/packages/loopx-codex-provider-routing/NATIVE.md @@ -19,10 +19,13 @@ windows, waiting cause and recovery. Tokens and identity claims stay private. restart permission, the provider requests a normal AppKit quit of only the inspected GPT instance, waits for exit and native idle, saves/switches the login, then reopens that same route. It never addresses a shared bundle id or quits a -different route. A refused quit, timeout, new process owner or remaining CLI/file -owner leaves credentials unchanged and pauses automatic retries until a check. +different route. With the separate force permission below, a refused or timed-out +normal quit can escalate against that same inspected instance. Otherwise those +failures, or remaining CLI/file owners, leave credentials unchanged and pause +automatic shutdowns until a check. Changed process ownership leaves credentials +unchanged and consumes the bounded retry budget. Without this additional permission it continues to wait for closed clients. -It does not implement hot login replacement, force termination, job replay or +It does not implement hot login replacement, job replay or resumption of an interrupted turn. If every account is unavailable, it waits for the observed reset and checks again. @@ -52,7 +55,11 @@ and reopen the GPT instance. Neither permission is enabled by installation. When enabled, normal refusal or timeout automatically force-quits only the observed GPT launch, then continues switching and reopening without another confirmation. The PID, executable, launch date, selection and quota freshness -must still match. Other CLI/file owners remain holds. Force quit can interrupt +must still match. The launch probe must explicitly contain `launch_time`: a +positive finite timestamp retains the launch-date fence; explicit `null` from +Launch Services uses the inspected PID and executable instead. Missing fields, +invalid types and non-finite timestamps cannot authorize force quit. A successful +normal quit never escalates. Other CLI/file owners remain holds. Force quit can interrupt work in that instance; it does not authorize replay of submitted jobs. An operator-owned service independent of the App can run the same observer: @@ -112,9 +119,12 @@ are distinct. The observer never buys credits or uses reset coupons. Only fresh active-account exhaustion starts automatic selection. Destinations must be allowlisted and freshly available. Selection follows the configured ring with a five-minute cooldown. An external native/manual choice wins and -starts a cooldown. Busy clients defer; refused/changed clients and remaining -file owners pause further shutdowns. Other repeated failures stop after three -attempts until an explicit check/reconnect. Resets require new observations. +starts a cooldown. Busy clients defer; refused/timed-out clients and remaining +file owners pause further shutdowns. Changed clients and other repeated failures +stop after three attempts until an explicit check/reconnect. Successful quota +reads do not reset that budget, which survives observer restart. A completed +switch, explicit check/reconfiguration or later native choice can reset it. +Quota resets require new observations. A paused observer still reads native selection metadata: a later native/manual choice invalidates the old account observations and failure pause, then starts the usual cooldown without another shutdown attempt. diff --git a/packages/loopx-codex-provider-routing/extension.toml b/packages/loopx-codex-provider-routing/extension.toml index ee2222e4b7..db4363e592 100644 --- a/packages/loopx-codex-provider-routing/extension.toml +++ b/packages/loopx-codex-provider-routing/extension.toml @@ -1,6 +1,6 @@ schema_version = "loopx_extension_manifest_v0" id = "loopx-codex-provider-routing" -version = "0.12.1" +version = "0.12.2" requires_loopx_api = ">=1,<2" permissions = [] diff --git a/packages/loopx-codex-provider-routing/pyproject.toml b/packages/loopx-codex-provider-routing/pyproject.toml index e6535ce34c..ef3acde379 100644 --- a/packages/loopx-codex-provider-routing/pyproject.toml +++ b/packages/loopx-codex-provider-routing/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "loopx-codex-provider-routing" -version = "0.12.1" +version = "0.12.2" description = "Native Codex account slots and exhausted-account handoff for LoopX" readme = "README.md" requires-python = ">=3.11" diff --git a/packages/loopx-codex-provider-routing/smoke/native_host_smoke.py b/packages/loopx-codex-provider-routing/smoke/native_host_smoke.py new file mode 100644 index 0000000000..9a4d5a51ea --- /dev/null +++ b/packages/loopx-codex-provider-routing/smoke/native_host_smoke.py @@ -0,0 +1,96 @@ +"""Offline smoke for native App shutdown during an exhausted-account handoff.""" + +from __future__ import annotations + +import importlib +import json +from pathlib import Path +import sys + +PACKAGE_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(PACKAGE_ROOT / "src")) + +native_host = importlib.import_module("loopx_codex_provider_routing.native_host") + +EXE = "/Applications/ChatGPT.app/Contents/MacOS/ChatGPT" + + +class Apps: + def __init__(self): + self.pid = 4242 + + def executable(self, _bundle): + return EXE + + def app_bundle(self, _spec, _role, _home): + return Path("/Applications/ChatGPT.app") + + def running(self, _exe, _spec): + return self.pid + + +def adapter(): + value = native_host.NativeHost.__new__(native_host.NativeHost) + value.apps = Apps() + value.spec = {} + value.home = Path("/tmp/native-host-smoke") + value.expected = "revision" + value.not_after = None + value.revision = lambda: "revision" + return value + + +def close_with(launch_time, *, force, refuse=False): + calls = [] + host = adapter() + + def fake_run(command, **kwargs): + calls.append((list(command), kwargs.get("input", ""))) + if kwargs["input"] == native_host.FORCE_QUIT or (kwargs["input"] == native_host.QUIT and not refuse): + host.apps.pid = None + + class Result: + stdout = json.dumps({"launch_time": launch_time, "accepted": not refuse}) + returncode = 0 + + return Result() + + original = native_host.subprocess.run + native_host.subprocess.run = fake_run + try: + closed = host.close_client("revision", force=force) + finally: + native_host.subprocess.run = original + return closed, calls + + +# An App started outside Launch Services reports a nil launchDate. The force +# path must still terminate it instead of failing the handoff as changed. +closed, calls = close_with(None, force=True, refuse=True) +assert closed is True +assert len(calls) == 3, calls +assert "launch_time" in calls[0][1] and "terminate" not in calls[0][1], calls[0] +assert "app.terminate" in calls[1][1] and "forceTerminate" not in calls[1][1], calls[1] +assert "forceTerminate" in calls[2][1], calls[2] +assert calls[2][0][-1] == "unknown", calls[2][0] + +# A registered App keeps the launch-time fence. +closed, calls = close_with(123.0, force=True, refuse=True) +assert closed is True +assert len(calls) == 3, calls +assert "forceTerminate" in calls[2][1], calls[2] +assert calls[2][0][-1] == "123.0", calls[2][0] + +# Force permission does not bypass a successful normal quit. +closed, calls = close_with(123.0, force=True) +assert closed is True +assert len(calls) == 2, calls +assert "app.terminate" in calls[1][1] and "forceTerminate" not in calls[1][1], calls[1] + +# The default path still asks the App to quit normally. +closed, calls = close_with(None, force=False) +assert closed is True +assert len(calls) == 1, calls +assert "terminate" in calls[0][1] and "forceTerminate" not in calls[0][1], calls[0] + +print("native host force handoff smoke passed") diff --git a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native.py b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native.py index 528564c1ef..e3025423fc 100644 --- a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native.py +++ b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native.py @@ -213,6 +213,10 @@ def tick(self, *, refresh: bool = False, manual_slot: str | None = None, return self._projection(state) if not refresh and not manual_slot and now < (state.get("retry_at") or 0): return self._projection(state) + if refresh or manual_slot: + # A deliberate check/choice starts a new retry budget; healthy + # quota alone does not prove that a handoff succeeded. + state["failure_count"] = 0 if "adapter" not in state: # An explicit check/manual choice can connect existing slots # without turning automatic effects on first. @@ -290,7 +294,7 @@ def tick(self, *, refresh: bool = False, manual_slot: str | None = None, **({"manual_slot": manual_slot} if manual_slot else {})}) state.update(observations=observed["observations"], accounts=decision["accounts"], last_checked_at=self.clock(), action=decision["action"], target=decision["target"], - retry_at=decision["retry_at"] or self.clock() + 60, code=None, failure_count=0) + retry_at=decision["retry_at"] or self.clock() + 60, code=None) if decision["action"] == "switch": if not manual_slot and not self.switch_allowed(): raise HostError("observer_stopped") @@ -329,7 +333,7 @@ def tick(self, *, refresh: bool = False, manual_slot: str | None = None, # Unrelated/remaining owners still require a safe-boundary wait. # Other repeated failures stop automatic attempts until Refresh. failures = state.get("failure_count", 0) + (code not in ("waiting_for_idle", "handoff_pending")) - if code in ("other_clients_busy", "client_quit_refused", "client_quit_timeout", "client_changed"): + if code in ("other_clients_busy", "client_quit_refused", "client_quit_timeout"): failures = 3 # Do not repeatedly close/reopen a refused client. state.update(action=code if code in ("waiting_for_idle", "selection_changed", "adapter_changed", "client_quit_refused", "client_quit_timeout", "client_changed", "handoff_pending") else "unavailable", code=code, retry_at=self.clock() + (5 if code == "handoff_pending" else min(300, 60 * 2 ** min(failures, 3))), failure_count=failures) diff --git a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_host.py b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_host.py index 0e3dfdefec..68b12dd6f5 100644 --- a/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_host.py +++ b/packages/loopx-codex-provider-routing/src/loopx_codex_provider_routing/native_host.py @@ -9,6 +9,7 @@ import hashlib import importlib import json +import math import os from pathlib import Path import selectors @@ -45,12 +46,20 @@ def __init__(self, code: str): if (!app || !app.executableURL || !app.isFinishedLaunching) throw Error("Client startup is still incomplete"); if (ObjC.unwrap(app.executableURL.path) !== argv[1]) throw Error("Client executable changed"); var started = Number(app.launchDate.timeIntervalSince1970); + if (argv.length > 2 && argv[2] !== "unknown" && started !== Number(argv[2])) throw Error("Client launch changed"); return JSON.stringify({launch_time: started, accepted: Boolean(app.terminate)}); }''' FORCE_QUIT = QUIT.replace( - 'var started = Number(app.launchDate.timeIntervalSince1970);\n return JSON.stringify({launch_time: started, accepted: Boolean(app.terminate)});', - 'if (Number(app.launchDate.timeIntervalSince1970) !== Number(argv[2])) throw Error("Client launch changed");\n if (!app.forceTerminate) throw Error("Client force quit failed");', + 'return JSON.stringify({launch_time: started, accepted: Boolean(app.terminate)});', + 'if (!app.forceTerminate) throw Error("Client force quit failed");', +) + +# Freeze launch identity before graceful quit so a refused/timed-out quit can +# escalate only against that same instance, without another confirmation. +PROBE = QUIT.replace( + 'return JSON.stringify({launch_time: started, accepted: Boolean(app.terminate)});', + 'return JSON.stringify({launch_time: started});', ) @@ -129,48 +138,57 @@ def close_client(self, expected: str, *, timeout: float = 25, force: bool = Fals raise HostError("selection_changed") if pid is None: return False - # Recheck exact ownership immediately before requesting a normal quit. + # Recheck exact ownership immediately before requesting a quit. if self.apps.running(exe, self.spec) != pid: raise HostError("client_changed") - try: - result = subprocess.run(["/usr/bin/osascript", "-l", "JavaScript", "-", str(pid), str(exe)], - input=QUIT, text=True, check=True, capture_output=True, timeout=10) - except (OSError, subprocess.SubprocessError): - raise HostError("client_quit_refused") from None - launch_time = None - accepted = True - if force or getattr(result, "stdout", ""): + if force: try: + result = subprocess.run(["/usr/bin/osascript", "-l", "JavaScript", "-", str(pid), str(exe)], + input=PROBE, text=True, check=True, capture_output=True, timeout=10) receipt = json.loads(result.stdout) - launch_time, accepted = receipt["launch_time"], receipt["accepted"] - if (type(accepted) is not bool or force - and (type(launch_time) not in (int, float) or launch_time <= 0)): + launch_time = receipt["launch_time"] + # Launch Services can report a nil launchDate for an App started + # outside the normal Finder/open registration path. Fall back to + # the inspected PID + executable identity instead of stalling. + if launch_time is not None and (type(launch_time) not in (int, float) + or not math.isfinite(launch_time) or launch_time <= 0): raise ValueError() - except (AttributeError, KeyError, TypeError, ValueError): - raise HostError("client_changed") from None - if not accepted and not force: - raise HostError("client_quit_refused") - deadline = time.monotonic() + timeout - while accepted and time.monotonic() < deadline: + except (OSError, subprocess.SubprocessError, AttributeError, KeyError, TypeError, ValueError, OverflowError): + if self.apps.running(exe, self.spec) is None: + return True + raise HostError("client_quit_refused") from None + if self.revision() != expected: + raise HostError("selection_changed") current = self.apps.running(exe, self.spec) if current is None: return True if current != pid: raise HostError("client_changed") - time.sleep(0.1) - if force: - # Force only this observed launch, under the same selection and - # fresh-quota fence. A recycled PID or a later manual choice wins. + if self.not_after is not None and time.time() >= self.not_after: + raise HostError("quota_stale") + identity = str(launch_time) if launch_time is not None else "unknown" + try: + return self._graceful_quit(pid, exe, timeout, identity=identity) + except HostError as error: + if error.code not in ("client_quit_refused", "client_quit_timeout"): + raise + # Opt-in covers fallback only; a quit dialog must not bypass a + # later selection/process/quota change before force termination. if self.revision() != expected: raise HostError("selection_changed") - if self.apps.running(exe, self.spec) != pid: + current = self.apps.running(exe, self.spec) + if current is None: + return True + if current != pid: raise HostError("client_changed") if self.not_after is not None and time.time() >= self.not_after: raise HostError("quota_stale") try: - subprocess.run(["/usr/bin/osascript", "-l", "JavaScript", "-", str(pid), str(exe), str(launch_time)], + subprocess.run(["/usr/bin/osascript", "-l", "JavaScript", "-", str(pid), str(exe), identity], input=FORCE_QUIT, text=True, check=True, capture_output=True, timeout=10) except (OSError, subprocess.SubprocessError): + if self.apps.running(exe, self.spec) is None: + return True raise HostError("client_quit_refused") from None deadline = time.monotonic() + 10 while time.monotonic() < deadline: @@ -180,6 +198,37 @@ def close_client(self, expected: str, *, timeout: float = 25, force: bool = Fals if current != pid: raise HostError("client_changed") time.sleep(0.1) + raise HostError("client_quit_timeout") + return self._graceful_quit(pid, exe, timeout) + + def _graceful_quit(self, pid: int, exe: Path, timeout: float, *, identity: str | None = None) -> bool: + command = ["/usr/bin/osascript", "-l", "JavaScript", "-", str(pid), str(exe)] + if identity is not None: + command.append(identity) + try: + result = subprocess.run(command, + input=QUIT, text=True, check=True, capture_output=True, timeout=10) + except (OSError, subprocess.SubprocessError): + raise HostError("client_quit_refused") from None + accepted = True + if getattr(result, "stdout", ""): + try: + receipt = json.loads(result.stdout) + accepted = receipt["accepted"] + if type(accepted) is not bool: + raise ValueError() + except (AttributeError, KeyError, TypeError, ValueError): + raise HostError("client_changed") from None + if not accepted: + raise HostError("client_quit_refused") + deadline = time.monotonic() + timeout + while accepted and time.monotonic() < deadline: + current = self.apps.running(exe, self.spec) + if current is None: + return True + if current != pid: + raise HostError("client_changed") + time.sleep(0.1) raise HostError("client_quit_timeout") def reopen(self, expected: str) -> dict: @@ -220,11 +269,11 @@ def switch(self, slot: str, expected: str, *, reopen: bool, restart_clients: boo raise HostError("other_clients_busy") from None time.sleep(0.2) result = self.manager.switch(slot) - except Exception as error: + except Exception: # A refused/failed transaction must not unnecessarily leave a # normally closed source App down. Native preflight blocks opening # an incomplete credential journal; a later manual choice wins. - if restart_clients and before.get("client_running") and getattr(error, "code", None) != "client_changed": + if restart_clients and before.get("client_running"): try: if self.status().get("client_running") is False: self.reopen(expected) diff --git a/tests/presentation/test_codex_accounts.py b/tests/presentation/test_codex_accounts.py index 12bd6db4c1..e5a5282e8b 100644 --- a/tests/presentation/test_codex_accounts.py +++ b/tests/presentation/test_codex_accounts.py @@ -50,8 +50,8 @@ def test_normal_quit_waits_for_owned_cold_app_registration(native, tmp_path, for ''') subprocess.run(["swiftc", str(source), "-framework", "Cocoa", "-o", str(executable)], check=True, capture_output=True, timeout=30) - host_type = importlib.import_module("loopx_codex_provider_routing.native_host").NativeHost - host = object.__new__(host_type) + host_module = importlib.import_module("loopx_codex_provider_routing.native_host") + host = object.__new__(host_module.NativeHost) host.home, host.spec = tmp_path, {} host.not_after = None host.revision = lambda: "fixture-selection" @@ -79,6 +79,15 @@ def running(*_): while not (tmp_path / "app.pid").exists() and time.monotonic() < deadline: time.sleep(.05) assert (tmp_path / "app.pid").exists() + # Challenge the real JXA instance fences before the authorized quit. + # Both graceful and forced paths must reject a different launch; + # force quit must also reject the wrong executable for this PID. + for script, path in [(host_module.QUIT, executable), (host_module.FORCE_QUIT, executable), + (host_module.FORCE_QUIT, executable.with_name("OtherFixture"))]: + with pytest.raises(subprocess.CalledProcessError): + subprocess.run(["/usr/bin/osascript", "-l", "JavaScript", "-", str(running()), str(path), "1"], + input=script, text=True, check=True, capture_output=True, timeout=10) + assert running() is not None assert host.close_client("fixture-selection", timeout=0.5 if force else 10, force=force) is True if force: assert running() is None @@ -307,6 +316,44 @@ def test_paused_unchanged_selection_does_not_repeat_usage_or_shutdown(native, tm assert provider._load()["failure_count"] == 3 +@pytest.mark.parametrize("force", [False, True]) +@pytest.mark.parametrize("code", ["client_changed", "native_unavailable"]) +def test_failed_handoffs_exhaust_a_durable_retry_budget(native, tmp_path, force, code): + provider, transport, clock = configured(native, tmp_path) + provider.configure(expected_revision=1, enabled=True, slots=["a", "b"], + restart_clients=True, force_quit_clients=force) + attempts = [] + failure = code + + def failing(adapter, operation, **parameters): + if operation == "switch": + attempts.append(parameters) + if failure: + raise native.HostError(failure) + return transport(adapter, operation, **parameters) + + counts = [] + for _ in range(4): + # Restart between due observations; the budget belongs to durable state. + provider = native.NativeAccounts(tmp_path, transport=failing, clock=clock) + provider.tick() + counts.append(provider._load()["failure_count"]) + clock.now += 1000 + assert counts == [1, 2, 3, 3] + assert len(attempts) == 3 and transport.active == "a" + assert all(attempt["force_quit_clients"] is force for attempt in attempts) + assert provider.tick(refresh=True)["active_slot"] == "a" + assert len(attempts) == 4 and provider._load()["failure_count"] == 1 + failure = None + clock.now += 1000 + assert provider.tick()["action"] == "selected" + assert provider._load()["failure_count"] == 0 + provider.configure(expected_revision=2, enabled=False, slots=["a", "b"]) + before = len(transport.calls), len(attempts) + assert provider.tick()["action"] == "disabled" + assert (len(transport.calls), len(attempts)) == before + + def test_lost_reply_recovers_readback_without_replaying_switch(native, tmp_path): provider, transport, clock = configured(native, tmp_path) transport.committed_without_reply = True @@ -522,9 +569,11 @@ def pending(adapter, operation, **parameters): assert result["policy"]["enabled"] is False and provider._load()["intent"] is None -def test_refused_client_is_not_repeatedly_closed_in_the_background(native, tmp_path): +@pytest.mark.parametrize("force", [False, True]) +def test_refused_client_is_not_repeatedly_closed_in_the_background(native, tmp_path, force): provider, transport, clock = configured(native, tmp_path) - provider.configure(expected_revision=1, enabled=True, slots=["a", "b"], restart_clients=True) + provider.configure(expected_revision=1, enabled=True, slots=["a", "b"], + restart_clients=True, force_quit_clients=force) calls = [] def refused(adapter, operation, **parameters): @@ -606,25 +655,30 @@ def quit(command, **parameters): return adapter, state, module -@pytest.mark.parametrize("fault", [None, "selection", "new-process", "stale-quota", "unreadable-launch"]) -def test_force_quit_requires_the_same_observed_launch_and_selection(native, monkeypatch, fault): +@pytest.mark.parametrize("fault,phase", [(None, "quit"), ("unreadable-launch", "probe"), + *[(fault, phase) for fault in ("selection", "new-process", "stale-quota") + for phase in ("probe", "quit")]]) +def test_force_quit_requires_the_same_observed_launch_and_selection(native, monkeypatch, fault, phase): adapter, state, module = client_host(native, monkeypatch) adapter.not_after = None forced = [] def stalled(command, **parameters): - if parameters["input"] == module.QUIT: + if parameters["input"] == (module.PROBE if phase == "probe" else module.QUIT): if fault == "selection": state["revision"] = "manual-choice" if fault == "new-process": state["pid"] = 33 if fault == "stale-quota": adapter.not_after = time.time() - 1 + if parameters["input"] == module.PROBE: return SimpleNamespace(stdout="{}" if fault == "unreadable-launch" else - json.dumps({"launch_time": 1000, "accepted": True})) + json.dumps({"launch_time": 1000})) + if parameters["input"] == module.QUIT: + return SimpleNamespace(stdout=json.dumps({"launch_time": 1000, "accepted": True})) forced.append(command) assert parameters["input"] == module.FORCE_QUIT - assert 'launchDate.timeIntervalSince1970) !== Number(argv[2])' in parameters["input"] + assert 'started !== Number(argv[2])' in parameters["input"] assert command[-3:] == ["11", "/synthetic/GPT.app/Contents/MacOS/GPT", "1000"] state["pid"] = None @@ -639,6 +693,53 @@ def stalled(command, **parameters): assert state["active"] == "a" and state["switch"] == 0 +@pytest.mark.parametrize("receipt", ["{}", "[]", "null", "not-json", '{"launch_time":true}', + '{"launch_time":"1000"}', '{"launch_time":0}', + '{"launch_time":-1}', '{"launch_time":NaN}', + '{"launch_time":Infinity}', '{"launch_time":-Infinity}']) +def test_incomplete_or_invalid_probe_cannot_authorize_any_quit(native, monkeypatch, receipt): + adapter, state, module = client_host(native, monkeypatch) + calls = [] + + def probe(command, **parameters): + calls.append(parameters["input"]) + assert parameters["input"] == module.PROBE + return SimpleNamespace(stdout=receipt) + + monkeypatch.setattr(module.subprocess, "run", probe) + with pytest.raises(module.HostError, match="client_quit_refused"): + adapter.close_client("initial", timeout=0, force=True) + assert calls == [module.PROBE] + assert state["pid"] == 11 and state["active"] == "a" and state["switch"] == 0 + + +@pytest.mark.parametrize("launch_time", [None, 1000.0]) +@pytest.mark.parametrize("outcome", ["graceful", "refused", "timeout"]) +def test_force_permission_only_escalates_a_failed_normal_quit(native, monkeypatch, launch_time, outcome): + adapter, state, module = client_host(native, monkeypatch) + adapter.not_after = None + calls = [] + + def quit(command, **parameters): + script = parameters["input"] + calls.append(script) + if script == module.PROBE: + return SimpleNamespace(stdout=json.dumps({"launch_time": launch_time})) + assert command[-1] == ("unknown" if launch_time is None else str(launch_time)) + if script == module.QUIT: + if outcome == "graceful": + state["pid"] = None + return SimpleNamespace(stdout=json.dumps({"accepted": outcome != "refused"})) + assert script == module.FORCE_QUIT and outcome != "graceful" + state["pid"] = None + + monkeypatch.setattr(module.subprocess, "run", quit) + assert adapter.close_client("initial", timeout=0 if outcome == "timeout" else 1, force=True) + expected = [module.PROBE, module.QUIT] + assert calls == (expected if outcome == "graceful" else [*expected, module.FORCE_QUIT]) + assert state["switch"] == 0 and state["active"] == "a" + + def test_force_permission_is_independent_and_survives_reconfiguration(native, tmp_path): provider, transport, _ = configured(native, tmp_path) assert provider._load()["policy"].get("force_quit_clients") is not True