From 9e9472bc665e914e00d8a1239e75919dac2a6895 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Fri, 23 Oct 2026 01:04:01 +0800
Subject: [PATCH 1/2] fix(native-accounts): force unattended App quit during
 handoff

Exhausted-account failover opted into force_quit_clients but still asked the
App to quit gracefully first, which surfaced the App's own quit confirmation.
The force branch also required a numeric NSRunningApplication launchDate; this
App reports nil, so every attempt raised client_changed before forceTerminate
ran. The switch then skipped reopening the App and the watcher latched
failure_count=3, stopping automatic retries until a manual refresh.

Probe the launch identity without quitting, force-terminate the inspected PID
directly when force is configured, and fall back to the PID + executable
identity when launchDate is unavailable. Reopen the App after a failed switch
when the selection is unchanged, and keep normal backoff for client_changed
instead of latching the watcher.

Verified with a new offline smoke (force path, numeric identity, normal quit),
the provider/recovery/operator smokes and the native policy node tests.
Bump the package to 0.12.2.

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .github/workflows/package-smoke.yml           |  1 +
 .../extension.toml                            |  2 +-
 .../pyproject.toml                            |  2 +-
 .../smoke/native_host_smoke.py                | 88 +++++++++++++++++++
 .../loopx_codex_provider_routing/native.py    |  2 +-
 .../native_host.py                            | 86 +++++++++++-------
 6 files changed, 148 insertions(+), 33 deletions(-)
 create mode 100644 packages/loopx-codex-provider-routing/smoke/native_host_smoke.py

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/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..5b9b605b27
--- /dev/null
+++ b/packages/loopx-codex-provider-routing/smoke/native_host_smoke.py
@@ -0,0 +1,88 @@
+"""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, pids):
+        self._pids = iter(pids)
+
+    def executable(self, _bundle):
+        return EXE
+
+    def app_bundle(self, _spec, _role, _home):
+        return Path("/Applications/ChatGPT.app")
+
+    def running(self, _exe, _spec):
+        return next(self._pids, None)
+
+
+def adapter(pids):
+    value = native_host.NativeHost.__new__(native_host.NativeHost)
+    value.apps = Apps(pids)
+    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):
+    calls = []
+
+    def fake_run(command, **kwargs):
+        calls.append((list(command), kwargs.get("input", "")))
+
+        class Result:
+            stdout = json.dumps({"launch_time": launch_time, "accepted": True})
+            returncode = 0
+
+        return Result()
+
+    original = native_host.subprocess.run
+    native_host.subprocess.run = fake_run
+    try:
+        # Ownership is checked before the quit, after the identity probe and
+        # once while waiting for the App root to exit.
+        closed = adapter([4242, 4242, 4242, None]).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)
+assert closed is True
+assert len(calls) == 2, calls
+assert "launch_time" in calls[0][1] and "terminate" not in calls[0][1], calls[0]
+assert "forceTerminate" in calls[1][1], calls[1]
+assert calls[1][0][-1] == "unknown", calls[1][0]
+
+# A registered App keeps the launch-time fence.
+closed, calls = close_with(123.0, force=True)
+assert closed is True
+assert len(calls) == 2, calls
+assert "forceTerminate" in calls[1][1], calls[1]
+assert calls[1][0][-1] == "123.0", calls[1][0]
+
+# 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..008ed61cd1 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
@@ -329,7 +329,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..47ccddb134 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
@@ -50,7 +50,14 @@ def __init__(self, code: str):
 
 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");',
+    'if (argv[2] !== "unknown" && Number(app.launchDate.timeIntervalSince1970) !== Number(argv[2])) throw Error("Client launch changed");\n  if (!app.forceTerminate) throw Error("Client force quit failed");',
+)
+
+# Read the launch identity without asking the App to quit. The force path uses
+# this so unattended failover never waits on the App's own quit confirmation.
+PROBE = QUIT.replace(
+    'return JSON.stringify({launch_time: started, accepted: Boolean(app.terminate)});',
+    'return JSON.stringify({launch_time: started});',
 )
 
 
@@ -129,48 +136,42 @@ 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:
+            # A graceful quit can surface the App's own confirmation dialog and
+            # stall an unattended handoff. Probe the launch identity and then
+            # force-terminate exactly this observed launch under the same fences.
             try:
-                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)):
+                result = subprocess.run(["/usr/bin/osascript", "-l", "JavaScript", "-", str(pid), str(exe)],
+                                        input=PROBE, text=True, check=True, capture_output=True, timeout=10)
+                launch_time = json.loads(result.stdout).get("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 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):
+                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.revision() != expected:
-                raise HostError("selection_changed")
-            if self.apps.running(exe, self.spec) != pid:
-                raise HostError("client_changed")
             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:
-                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 +181,31 @@ 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")
+        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
+        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:
@@ -224,7 +250,7 @@ def switch(self, slot: str, expected: str, *, reopen: bool, restart_clients: boo
             # 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)

From ada2d176c441fcbe8c8f630b2136528c783d09aa Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Fri, 23 Oct 2026 11:16:13 +0800
Subject: [PATCH 2/2] fix(native-accounts): bound retries and preserve graceful
 quit permission

Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
 .../loopx-codex-provider-routing/NATIVE.md    |  24 ++--
 .../smoke/native_host_smoke.py                |  40 +++---
 .../loopx_codex_provider_routing/native.py    |   6 +-
 .../native_host.py                            |  47 +++++--
 tests/presentation/test_codex_accounts.py     | 119 ++++++++++++++++--
 5 files changed, 191 insertions(+), 45 deletions(-)

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/smoke/native_host_smoke.py b/packages/loopx-codex-provider-routing/smoke/native_host_smoke.py
index 5b9b605b27..9a4d5a51ea 100644
--- a/packages/loopx-codex-provider-routing/smoke/native_host_smoke.py
+++ b/packages/loopx-codex-provider-routing/smoke/native_host_smoke.py
@@ -16,8 +16,8 @@
 
 
 class Apps:
-    def __init__(self, pids):
-        self._pids = iter(pids)
+    def __init__(self):
+        self.pid = 4242
 
     def executable(self, _bundle):
         return EXE
@@ -26,12 +26,12 @@ def app_bundle(self, _spec, _role, _home):
         return Path("/Applications/ChatGPT.app")
 
     def running(self, _exe, _spec):
-        return next(self._pids, None)
+        return self.pid
 
 
-def adapter(pids):
+def adapter():
     value = native_host.NativeHost.__new__(native_host.NativeHost)
-    value.apps = Apps(pids)
+    value.apps = Apps()
     value.spec = {}
     value.home = Path("/tmp/native-host-smoke")
     value.expected = "revision"
@@ -40,14 +40,17 @@ def adapter(pids):
     return value
 
 
-def close_with(launch_time, *, force):
+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": True})
+            stdout = json.dumps({"launch_time": launch_time, "accepted": not refuse})
             returncode = 0
 
         return Result()
@@ -55,9 +58,7 @@ class Result:
     original = native_host.subprocess.run
     native_host.subprocess.run = fake_run
     try:
-        # Ownership is checked before the quit, after the identity probe and
-        # once while waiting for the App root to exit.
-        closed = adapter([4242, 4242, 4242, None]).close_client("revision", force=force)
+        closed = host.close_client("revision", force=force)
     finally:
         native_host.subprocess.run = original
     return closed, calls
@@ -65,19 +66,26 @@ class Result:
 
 # 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)
+closed, calls = close_with(None, force=True, refuse=True)
 assert closed is True
-assert len(calls) == 2, calls
+assert len(calls) == 3, calls
 assert "launch_time" in calls[0][1] and "terminate" not in calls[0][1], calls[0]
-assert "forceTerminate" in calls[1][1], calls[1]
-assert calls[1][0][-1] == "unknown", calls[1][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 "forceTerminate" in calls[1][1], calls[1]
-assert calls[1][0][-1] == "123.0", calls[1][0]
+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)
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 008ed61cd1..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")
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 47ccddb134..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,16 +46,17 @@ 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 (argv[2] !== "unknown" && 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");',
 )
 
-# Read the launch identity without asking the App to quit. The force path uses
-# this so unattended failover never waits on the App's own quit confirmation.
+# 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});',
@@ -140,19 +142,18 @@ def close_client(self, expected: str, *, timeout: float = 25, force: bool = Fals
         if self.apps.running(exe, self.spec) != pid:
             raise HostError("client_changed")
         if force:
-            # A graceful quit can surface the App's own confirmation dialog and
-            # stall an unattended handoff. Probe the launch identity and then
-            # force-terminate exactly this observed launch under the same fences.
             try:
                 result = subprocess.run(["/usr/bin/osascript", "-l", "JavaScript", "-", str(pid), str(exe)],
                                         input=PROBE, text=True, check=True, capture_output=True, timeout=10)
-                launch_time = json.loads(result.stdout).get("launch_time")
+                receipt = json.loads(result.stdout)
+                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 launch_time <= 0):
+                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 (OSError, subprocess.SubprocessError, AttributeError, KeyError, TypeError, ValueError):
+            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
@@ -166,6 +167,22 @@ def close_client(self, expected: str, *, timeout: float = 25, force: bool = Fals
             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")
+            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), identity],
                                input=FORCE_QUIT, text=True, check=True, capture_output=True, timeout=10)
@@ -182,8 +199,14 @@ def close_client(self, expected: str, *, timeout: float = 25, force: bool = Fals
                     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(["/usr/bin/osascript", "-l", "JavaScript", "-", str(pid), str(exe)],
+            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
@@ -246,7 +269,7 @@ 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.
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