Unbind model rules automatically when deleting a bound credential

app/admin_api.py

@@ -42,6 +42,15 @@ async def _body(request, maximum=65536, *, allow_empty=False):
     return value
 
 
+def _unbind_requested(query: str) -> bool:
+    """Mirror the delete route's Boolean query parsing for the unbind confirmation flag."""
+    for pair in (query or "").split("&"):
+        key, _, value = pair.partition("=")
+        if key == "unbind" and value.lower() in ("1", "true", "on", "yes"):
+            return True
+    return False
+
+
 def _public_credential(item):
     # Never pass arbitrary credential manager fields through to the browser.
     fields = {"id", "account_key", "name", "filename", "enabled", "label", "profile", "region", "site",
@@ -73,6 +82,8 @@ def install_admin(app, config, gateway):
     # otherwise never clear a snapshot belonging to a superseded key.
     auth.reconcile()
     mutation_lock = threading.RLock()
+    # The confirmed credential delete shares this lock so rule edits cannot interleave.
+    config["admin_mutation_lock"] = mutation_lock
     oauth_lock = threading.RLock()
     oauth_tasks = OrderedDict()
 
@@ -173,10 +184,11 @@ async def dispatch(request):
             name = path.removeprefix("/admin/credentials/")
             if not _valid_name(name):
                 return error_response(400, "凭证文件名无效")
-            try:
-                await run_in_threadpool(gateway.admin_delete_guard, name)
-            except (ValueError, HTTPException):
-                return error_response(409, "凭证仍被模型策略引用，请先移除绑定")
+            if not _unbind_requested(request.url.query):
+                try:
+                    await run_in_threadpool(gateway.admin_delete_guard, name)
+                except (ValueError, HTTPException):
+                    return error_response(409, "凭证仍被模型策略引用，请先移除绑定")
         return None
 
     # Middleware is deliberately installed only here, never on module import.

app/control_store.py

@@ -193,6 +193,43 @@ def set_credential(self, account_key, enabled):
             raise ValueError("enabled 必须为布尔值")
         return self._update(None, lambda state: state["credentials"].setdefault(account_key, {}).update(enabled=enabled))
 
+    def unbind_credential(self, account_key):
+        """Drop an account key from every rule, returning {source: old_ids} for rollback."""
+        _identifier(account_key, "账号指纹")
+        if not any(account_key in (rule.get("credential_ids") or [])
+                   for rule in self.snapshot()["models"].values()):
+            return {}  # Nothing references the identity; keep the revision stable.
+        affected = {}
+        def change(state):
+            for source, rule in state["models"].items():
+                ids = rule.get("credential_ids") or []
+                if account_key in ids:
+                    affected[source] = list(ids)
+                    rule["credential_ids"] = [identity for identity in ids if identity != account_key]
+        self._update(None, change)
+        return affected
+
+    def restore_bindings(self, account_key, affected):
+        """Re-add the identity where a rule still matches its post-unbind state.
+
+        Concurrent edits that changed a rule's binding list are preserved.
+        """
+        _identifier(account_key, "账号指纹")
+        pending = {source: ids for source, ids in dict(affected or {}).items() if isinstance(ids, list)}
+        def restorable(rule, ids):
+            if rule is None:
+                return False  # A rule deleted meanwhile stays deleted.
+            current = rule.get("credential_ids") or []
+            return current == [identity for identity in ids if identity != account_key]
+        snapshot = self.snapshot()
+        if not any(restorable(snapshot["models"].get(source), ids) for source, ids in pending.items()):
+            return snapshot  # Nothing to restore; concurrent edits win.
+        def change(state):
+            for source, ids in pending.items():
+                if restorable(state["models"].get(source), ids):
+                    state["models"][source]["credential_ids"] = list(ids)
+        return self._update(None, change)
+
     def set_auto_checkin(self, account_key, enabled):
         _identifier(account_key, "账号指纹")
         if type(enabled) is not bool:

app/gateway_management.py

@@ -109,12 +109,29 @@ def admin_set_auto_travel(self, identity, enabled):
                 raise HTTPException(400, "旅行仅适用于国内账号")
             self.CONFIG["control_store"].set_auto_travel(identity, enabled)
 
-    def admin_delete_guard(self, name):
+    def admin_delete_guard(self, name, *, unbind=False):
+        """Block a still-bound delete, or return the identity to unbind after removal."""
         rows = self.admin_credential_inventory()
         row = next((row for row in rows if row["name"] == name or row["id"] == name), None)
-        if row and row["bindings"]:
+        if row is None or not row["bindings"]:
+            return None
+        if not unbind:
             raise HTTPException(status_code=409, detail={"message": "凭证仍被模型规则引用，请先移除绑定",
                                                         "models": row["bindings"]})
+        return row["id"]
+
+    def admin_unbind_credential(self, identity):
+        """Unbind now and return the rollback record used when removal fails."""
+        control = self.CONFIG.get("control_store")
+        if control is None:
+            return {}
+        return {"identity": identity, "rules": control.unbind_credential(identity)}
+
+    def admin_restore_bindings(self, rollback):
+        """Put bindings back after a failed removal so routing never widens silently."""
+        control = self.CONFIG.get("control_store")
+        if control is not None:
+            control.restore_bindings(rollback["identity"], rollback["rules"])
 
     def admin_model_inventory(self):
         pool = self.CONFIG.get("cred_pool")

converter.py

@@ -2085,15 +2085,29 @@ async def admin_add_credential(request: Request,
 
 @app.delete("/admin/credentials/{name}")
 def admin_del_credential(name: str,
+                         unbind: bool = False,
                          authorization: Optional[str] = Header(default=None),
                          x_api_key: Optional[str] = Header(default=None, alias="X-Api-Key")):
-    """Delete the named .info file and remove its credential from the pool."""
+    """Delete the named .info file, unbinding model rules only after removal succeeds."""
     _check_admin_auth(authorization, x_api_key)
     pool = CONFIG.get("cred_pool")
-    if CONFIG.get("management") is not None:
-        CONFIG["management"].admin_delete_guard(os.path.basename(name))
-    if pool is None or not pool.remove_file(os.path.basename(name)):
-        raise HTTPException(status_code=404, detail={"error": {"message": f"凭据不在池中: {name}", "type": "invalid_request_error"}})
+    rollback = {}
+    identity = None
+    management = CONFIG.get("management")
+    # Serialize with model-rule edits so a rebind cannot slip between unbind and removal.
+    with CONFIG.get("admin_mutation_lock") or nullcontext():
+        if management is not None:
+            identity = management.admin_delete_guard(os.path.basename(name), unbind=unbind)
+            if identity is not None:
+                rollback = management.admin_unbind_credential(identity)
+        if pool is None or not pool.remove_file(os.path.basename(name)):
+            # A failed removal rolls the unbind back so routing never widens silently.
+            if management is not None and rollback:
+                management.admin_restore_bindings(rollback)
+            raise HTTPException(status_code=404, detail={"error": {"message": f"凭据不在池中: {name}", "type": "invalid_request_error"}})
+        if identity is not None and management is not None:
+            # Sweep any rebind that raced the removal from outside the shared lock.
+            management.admin_unbind_credential(identity)
     return {"removed": os.path.basename(name)}
 
 

tests/test_admin_api.py

@@ -511,6 +511,13 @@ def test_inventory_toggle_and_delete_guard(self):
         self.gateway.admin_set_credential_enabled.assert_called_once_with("fingerprint", False)
         self.gateway.admin_delete_guard.side_effect = ValueError("referenced")
         self.assertEqual(self.client.delete("/admin/credentials/first.info", headers=self.headers).status_code, 409)
+        # The confirmed unbind flow bypasses the middleware guard; the delete route owns it.
+        self.gateway.admin_delete_guard.reset_mock()
+        self.gateway.admin_delete_guard.side_effect = None
+        self.assertEqual(
+            self.client.delete("/admin/credentials/first.info?unbind=1", headers=self.headers).status_code,
+            200)
+        self.gateway.admin_delete_guard.assert_not_called()
 
     def test_upload_limits_paths_and_secret_errors(self):
         for files in ([{"name": "../first.info", "content": "{}"}], [{"name": "db.sqlite3", "content": "{}"}],

tests/test_control_store.py

@@ -82,6 +82,40 @@ def test_legacy_combined_scopes_load_without_widening_and_require_explicit_edit(
         self.assertEqual(reopened.snapshot()["revision"], 1)
 
 
+    def test_unbind_credential_removes_only_the_target_identity(self):
+        self.store.update_model("real", {"public_id": "public", "credential_ids": ["fingerprint", "other"]}, 0)
+        self.store.update_model("second", {"credential_ids": ["fingerprint"]}, 1)
+        self.store.update_model("third", {"credential_ids": ["other"]}, 2)
+        affected = self.store.unbind_credential("fingerprint")
+        self.assertEqual(affected, {"real": ["fingerprint", "other"], "second": ["fingerprint"]})
+        self.assertEqual(self.store.snapshot()["revision"], 4)  # Three setup bumps, then one for the unbind.
+        self.assertEqual(self.store.snapshot()["models"]["real"]["credential_ids"], ["other"])
+        self.assertEqual(self.store.snapshot()["models"]["second"]["credential_ids"], [])
+        self.assertEqual(self.store.snapshot()["models"]["third"]["credential_ids"], ["other"])
+        self.assertEqual(self.store.unbind_credential("unknown-identity"), {})  # No-op keeps the revision.
+        self.assertEqual(self.store.snapshot()["revision"], 4)
+        self.store.restore_bindings("fingerprint", affected)  # Rollback restores both rules in one bump.
+        self.assertEqual(self.store.snapshot()["models"]["real"]["credential_ids"], ["fingerprint", "other"])
+        self.assertEqual(self.store.snapshot()["models"]["second"]["credential_ids"], ["fingerprint"])
+        self.assertEqual(self.store.snapshot()["revision"], 5)
+        self.store.restore_bindings("fingerprint", affected)  # An already-restored state bumps nothing.
+        self.assertEqual(self.store.snapshot()["revision"], 5)
+
+    def test_rollback_keeps_concurrent_edits_and_deleted_rules(self):
+        self.store.update_model("real", {"credential_ids": ["fingerprint", "other"]}, 0)
+        self.store.update_model("gone", {"custom": True, "public_id": "gone-public", "credential_ids": ["fingerprint"]}, 1)
+        affected = self.store.unbind_credential("fingerprint")
+        revision = self.store.snapshot()["revision"]
+        # Concurrent edits: one rule rebound elsewhere, another deleted outright.
+        self.store.update_model("real", {"credential_ids": ["fresh"]}, revision)
+        self.store.delete_model("gone", self.store.snapshot()["revision"])
+        self.store.restore_bindings("fingerprint", affected)
+        self.assertEqual(self.store.snapshot()["models"]["real"]["credential_ids"], ["fresh"])
+        self.assertNotIn("gone", self.store.snapshot()["models"])
+        with self.assertRaises(ValueError):
+            self.store.unbind_credential("not a fingerprint")
+
+
     def test_credential_metadata_and_no_secret_settings(self):
         self.store.set_credential("account-fingerprint", False)
         self.assertEqual(self.store.snapshot()["credentials"], {"account-fingerprint": {"enabled": False}})

tests/test_credential_unbind.py

@@ -0,0 +1,118 @@
+"""Deleting credentials that are still bound by model rules must auto-unbind on confirmation."""
+import sys
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))  # Allow direct execution.
+
+import json
+import os
+import tempfile
+import time
+import unittest
+from types import SimpleNamespace
+from unittest.mock import patch
+import threading
+
+
+from fastapi import HTTPException
+from fastapi.testclient import TestClient
+
+import converter
+from app.control_store import ControlStore
+from app.gateway_management import Management
+
+
+class UnbindDeleteTests(unittest.TestCase):
+    def setUp(self):
+        root = Path(self.enterContext(tempfile.TemporaryDirectory()))
+        self.enterContext(patch.dict(os.environ, {"CODEBUDDY_AUTH_DIR": str(root)}))
+        self.enterContext(patch.dict(converter.CONFIG, {"log_path": None, "cred_pool": None, "cred": None}))
+        self.path = root / "account.info"
+        self.path.write_text(json.dumps({"account": {"uid": "synthetic"}, "auth": {
+            "accessToken": "old-synthetic-token", "refreshToken": "synthetic-refresh",
+            "domain": "www.codebuddy.cn", "expiresAt": (time.time() + 86400) * 1000,
+            "lastRefreshTime": time.time() * 1000}}), encoding="utf-8")
+        self.pool = converter.CredentialPool([self.path])
+        self.store = ControlStore(root / "control.sqlite3")
+        self.addCleanup(self.store.close)
+        self.config = {"cred_pool": self.pool, "control_store": self.store,
+                       "ledger": None, "trial_ledger": None}
+        self.management = Management(SimpleNamespace(CONFIG=self.config))
+        self.identity = self.pool.entries()[0]["account_key"]
+
+    def bind(self):
+        self.store.update_model("glm-4-flash", {"credential_ids": [self.identity]}, 0)
+
+    def test_bound_delete_is_blocked_without_the_confirmation_flag(self):
+        self.bind()
+        with self.assertRaises(HTTPException) as ctx:
+            self.management.admin_delete_guard("account.info")
+        self.assertEqual(ctx.exception.status_code, 409)
+        self.assertEqual(ctx.exception.detail["models"], ["glm-4-flash"])
+
+    def test_confirmed_unbind_rolls_back_when_removal_fails(self):
+        self.store.update_model("glm-4-flash", {"public_id": "glm4", "credential_ids": [self.identity, "other"]}, 0)
+        identity = self.management.admin_delete_guard("account.info", unbind=True)
+        self.assertEqual(identity, self.identity)
+        rollback = self.management.admin_unbind_credential(identity)
+        self.assertEqual(rollback, {"identity": self.identity, "rules": {"glm-4-flash": [self.identity, "other"]}})
+        self.assertEqual(self.store.snapshot()["models"]["glm-4-flash"]["credential_ids"], ["other"])
+        self.management.admin_restore_bindings(rollback)  # A failed removal undoes the unbind.
+        rule = self.store.snapshot()["models"]["glm-4-flash"]
+        self.assertEqual(rule["credential_ids"], [self.identity, "other"])
+        self.assertEqual(rule["public_id"], "glm4")
+        self.assertTrue(self.path.exists())
+
+    def test_rollback_preserves_concurrent_binding_edits(self):
+        self.store.update_model("glm-4-flash", {"credential_ids": [self.identity, "other"]}, 0)
+        identity = self.management.admin_delete_guard("account.info", unbind=True)
+        rollback = self.management.admin_unbind_credential(identity)
+        # Another administrator rebinds the rule between the unbind and the rollback.
+        revision = self.store.snapshot()["revision"]
+        self.store.update_model("glm-4-flash", {"credential_ids": ["fresh"]}, revision)
+        self.management.admin_restore_bindings(rollback)
+        self.assertEqual(self.store.snapshot()["models"]["glm-4-flash"]["credential_ids"], ["fresh"])
+
+    def test_route_keeps_bindings_when_removal_fails(self):
+        self.bind()
+        config = dict(self.config, api_key="secret", management=self.management)
+        self.enterContext(patch.dict(converter.CONFIG, config))
+        client = TestClient(converter.app)
+        headers = {"X-Api-Key": "secret"}
+        with patch.object(self.pool, "remove_file", return_value=False):
+            response = client.delete("/admin/credentials/account.info?unbind=1", headers=headers)
+            self.assertEqual(response.status_code, 404, response.text)
+        self.assertEqual(self.store.snapshot()["models"]["glm-4-flash"]["credential_ids"],
+                         [self.identity])  # A failed removal never widens routing to automatic.
+        response = client.delete("/admin/credentials/account.info?unbind=1", headers=headers)
+        self.assertEqual(response.status_code, 200, response.text)
+        self.assertEqual(self.store.snapshot()["models"]["glm-4-flash"]["credential_ids"], [])
+        self.assertFalse(self.path.exists())
+
+    def test_route_sweeps_a_rebind_that_raced_the_removal(self):
+        self.bind()
+        config = dict(self.config, api_key="secret", management=self.management,
+                      admin_mutation_lock=threading.RLock())
+        self.enterContext(patch.dict(converter.CONFIG, config))
+        client = TestClient(converter.app)
+        headers = {"X-Api-Key": "secret"}
+        real_remove = self.pool.remove_file
+        def rebind_during_removal(file_name):
+            # A writer outside the shared lock rebinds while the credential is pooled.
+            revision = self.store.snapshot()["revision"]
+            self.store.update_model("glm-4-flash", {"credential_ids": [self.identity]}, revision)
+            return real_remove(file_name)
+        with patch.object(self.pool, "remove_file", side_effect=rebind_during_removal):
+            response = client.delete("/admin/credentials/account.info?unbind=1", headers=headers)
+            self.assertEqual(response.status_code, 200, response.text)
+        self.assertEqual(self.store.snapshot()["models"]["glm-4-flash"]["credential_ids"], [])
+        self.assertFalse(self.path.exists())
+
+    def test_unbound_and_unknown_credentials_stay_on_the_fast_path(self):
+        self.assertIsNone(self.management.admin_delete_guard("account.info"))  # No bindings.
+        self.assertIsNone(self.management.admin_delete_guard("missing.info", unbind=True))  # Pool reports 404.
+        self.assertEqual(self.store.snapshot()["revision"], 0)
+
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

web/e2e/admin.spec.ts

@@ -264,6 +264,26 @@ test("credential OAuth terminates, upload/export and safe-name deletion are wire
     )
     .toBe(true);
 });
+test("deleting a bound credential confirms the automatic unbind", async ({ page }) => {
+  const mock = await mockAPI(page);
+  await page.route("**/admin/credentials", async (route) => {
+    if (route.request().method() !== "GET") return route.fallback();
+    return route.fulfill({ json: { credentials: [{ ...credential, bindings: ["mock-model"] }] } });
+  });
+  await page.goto("/dashboard/credentials");
+  await page.getByRole("button", { name: "删除", exact: true }).click();
+  await expect(page.getByText("该凭证仍被模型规则引用：mock-model。")).toBeVisible();
+  await expect(page.getByText("确认后将自动解除这些绑定，规则回退为自动选择账号。")).toBeVisible();
+  await page.getByRole("button", { name: "解除绑定并删除凭证" }).click();
+  await expect
+    .poll(() =>
+      mock.calls.some(
+        (call) =>
+          call.method === "DELETE" && call.path === "/admin/credentials/mock-account.info?unbind=1",
+      ),
+    )
+    .toBe(true);
+});
 test("international OAuth choices send distinct sites and show the selected official host", async ({
   page,
 }) => {