loopx-agent · Nov 4, 2026 · Nov 4, 2026 · Nov 4, 2026 · Nov 4, 2026
@@ -124,6 +124,14 @@ 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.
 
+Version 0.14.1 reconciles selected-slot metadata when the current login uniquely
+matches an existing slot and its configured provider route agrees. Status and
+launch preflight use the existing account locks; a pending transaction, occupied
+lock, ambiguous identity or inconsistent route retains the old selection.
+An invalid or unreadable slot is not skipped to manufacture a unique match.
+This reconciliation never rewrites credentials or configuration, and does not
+prove that an already running client has reloaded its login.
+
 ## Prerequisites and activation
 
 Install in the same Python environment as LoopX, then restart its App/Chat service:

@@ -1,6 +1,6 @@
 schema_version = "loopx_extension_manifest_v0"
 id = "loopx-codex-provider-routing"
-version = "0.14.0"
+version = "0.14.1"
 requires_loopx_api = ">=1,<2"
 permissions = []
 

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "loopx-codex-provider-routing"
-version = "0.14.0"
+version = "0.14.1"
 description = "Native Codex account slots and exhausted-account handoff for LoopX"
 readme = "README.md"
 requires-python = ">=3.11"

@@ -330,13 +330,58 @@ def selected(self):
         state = self.state()
         if state:
             raw = read_private(self.auth)
-            if identity(raw) != state['identity']:
+            current = identity(raw)
+            if current != state['identity']:
+                adopted = self._adopt(current)
+                if adopted is not None:
+                    return adopted, raw
                 raise ValueError('GPT login changed outside the slot manager; inspect before overwriting any account')
             return state, raw
         if self.auth.exists():
+            raw = read_private(self.auth)
+            adopted = self._adopt(identity(raw))
+            if adopted is not None:
+                return adopted, raw
             raise ValueError('Register the current GPT login first: codex app enroll a --current')
         return None, None
 
+    def _adopt(self, current):
+        """Adopt the home login when it matches exactly one stored slot.
+
+        An operator or another tool can restore a slot's credentials directly,
+        for example when returning from the API route after an outage. Only a
+        unique identity match is adopted; an unknown or ambiguous login still
+        requires inspection.
+        """
+        folder = checked(self.root / 'slots')
+        if not folder.is_dir() or self.pending.exists():
+            return None
+        try:
+            # Metadata adoption must not race a credential transaction. Shared
+            # credential access does not rewrite or refresh the current login.
+            with self.lock(enrollment=True):
+                raw = read_private(self.auth)
+                if self.pending.exists() or identity(raw) != current:
+                    return None
+                config = self.config.read_bytes() if self.config.exists() else b''
+                expected = tomllib.loads(file_config(config, raw).decode())
+                actual = tomllib.loads(config.decode())
+                if (actual.get('model_provider', 'openai') != expected.get('model_provider', 'openai')
+                        or actual.get('openai_base_url') != expected.get('openai_base_url')
+                        or actual.get('forced_login_method') not in (None, expected['forced_login_method'])):
+                    return None
+                # Use the same complete, validated slot inventory as status;
+                # an unreadable slot is not proof that the match is unique.
+                matches = [entry.name for entry in sorted(folder.iterdir())
+                           if identity(read_private(self.slot(entry.name))) == current]
+                if len(matches) != 1:
+                    return None
+                state = {'slot': matches[0], 'identity': current, 'binding': self.binding}
+                atomic(self.active, encoded(state))
+                return state
+        except BlockingIOError:
+            return None  # The existing owner retries after its lock releases.
+
     def targets(self, key):
         if key in ('auth', 'config', 'active', 'binding'):
             return checked({'auth': self.auth, 'config': self.config, 'active': self.active,
@@ -426,11 +471,14 @@ def status(self):
                 kinds[child.name] = auth_kind(read_private(self.slot(child.name)))
         state = self.state()
         matches = None
-        if state and self.auth.exists():
+        if self.auth.exists():
             try:
-                matches = identity(read_private(self.auth)) == state['identity']
+                current = identity(read_private(self.auth))
+                if not state or current != state['identity']:
+                    state = self._adopt(current) or state
+                matches = current == state['identity'] if state else None
             except ValueError:
-                matches = False
+                matches = False if state else None
         return {'slots': names, 'slot_kinds': kinds, 'active_slot': state['slot'] if state else None,
                 'current_login_matches': matches, 'recovery_required': self.pending.exists(),
                 'home': str(self.spec['home']), 'frontend': str(self.spec['frontend']),

@@ -141,6 +141,82 @@ def test_current_registration_preserves_auth_model_and_comments(self):
         for path in self.manager.root.rglob('*'):
             self.assertEqual(path.stat().st_mode & 0o777, 0o700 if path.is_dir() else 0o600)
 
+    def test_selected_adopts_unique_external_slot_login(self):
+        self.ready()
+        self.assertEqual(self.manager.status()['active_slot'], 'a')
+        # An operator or another tool restores slot b's vault entry directly.
+        accounts.atomic(self.manager.auth, self.b)
+        self.assertEqual(self.manager.status()['active_slot'], 'b')
+
+    def test_adoption_waits_for_an_interrupted_transaction(self):
+        self.ready()
+        accounts.atomic(self.manager.auth, self.b)
+        before = self.manager.active.read_bytes()
+        accounts.atomic(self.manager.pending, b'{}\n')
+        status = self.manager.status()
+        self.assertTrue(status['recovery_required'])
+        self.assertFalse(status['current_login_matches'])
+        self.assertEqual(self.manager.active.read_bytes(), before)
+
+    def test_adoption_resumes_after_the_account_lock_releases(self):
+        self.ready()
+        accounts.atomic(self.manager.auth, self.b)
+        before = self.manager.active.read_bytes()
+        with self.manager.lock():
+            self.assertFalse(self.manager.status()['current_login_matches'])
+            self.assertEqual(self.manager.active.read_bytes(), before)
+        self.assertEqual(self.manager.status()['active_slot'], 'b')
+        self.assertEqual(self.manager.auth.read_bytes(), self.b)
+
+    def test_adoption_requires_the_restored_login_and_route_to_agree(self):
+        self.ready()
+        subscription_config = self.manager.config.read_bytes()
+        self.manager.enroll_api('openai-api', api_key='sk-loopx-relay-synthetic-local-token',
+                                base_url='http://127.0.0.1:32100/v1')
+        self.manager.switch('openai-api')
+        accounts.atomic(self.manager.auth, self.b)
+        before = self.manager.active.read_bytes(), self.manager.config.read_bytes()
+        self.assertFalse(self.manager.status()['current_login_matches'])
+        self.assertEqual((self.manager.active.read_bytes(), self.manager.config.read_bytes()), before)
+        # The external tool finishes restoring its matching subscription route.
+        accounts.atomic(self.manager.config, subscription_config)
+        self.assertEqual(self.manager.status()['active_slot'], 'b')
+
+    def test_status_recognizes_a_known_login_without_an_active_annotation(self):
+        self.ready()
+        accounts.atomic(self.manager.auth, self.b)
+        self.manager.active.unlink()
+        self.assertEqual(self.manager.status()['active_slot'], 'b')
+        self.assertTrue(self.manager.status()['current_login_matches'])
+
+    def test_adoption_keeps_an_ambiguous_identity_unselected(self):
+        self.ready()
+        accounts.atomic(self.manager.slot('c'), self.b)
+        accounts.atomic(self.manager.auth, self.b)
+        before = self.manager.active.read_bytes()
+        self.assertFalse(self.manager.status()['current_login_matches'])
+        self.assertEqual(self.manager.active.read_bytes(), before)
+
+    def test_adoption_does_not_skip_an_unreadable_slot(self):
+        self.ready()
+        accounts.atomic(self.manager.slot('c'), b'{}\n')
+        accounts.atomic(self.manager.auth, self.b)
+        before = self.manager.active.read_bytes()
+        with self.assertRaisesRegex(ValueError, 'Invalid native account cache'):
+            self.manager.selected()
+        self.assertEqual(self.manager.active.read_bytes(), before)
+
+    def test_unregistered_status_remains_unknown_without_creating_a_vault(self):
+        direct = accounts.encoded({'auth_mode': 'apikey', 'OPENAI_API_KEY': 'synthetic-external-key'})
+        for raw in (self.a, direct, b'null\n'):
+            with self.subTest(raw=raw):
+                accounts.atomic(self.manager.auth, raw)
+                before = {p: p.read_bytes() for p in self.home.rglob('*') if p.is_file()}
+                status = self.manager.status()
+                self.assertIsNone(status['active_slot'])
+                self.assertIsNone(status['current_login_matches'])
+                self.assertEqual({p: p.read_bytes() for p in self.home.rglob('*') if p.is_file()}, before)
+
     def test_native_enrollment_does_not_select_or_change_current_login(self):
         self.current()
         result = self.enroll_b()